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

Implemented inbound and outbound channel adapter hierarchy, MessageHandlerAdapter, and a parameterized SimpleMethodInvoker

Browse Source
This commit is contained in:
Mark Fisher
2007-12-10 05:11:19 +00:00
parent 0ababb60fe
commit 39d31e087e
25 changed files with 857 additions and 314 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
spring-eai-core/src/main/java/org/springframework/integration/MessageSource.java
View File

@@ -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
* <code>null</code> 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
* <code>null</code> if the specified timeout period
* elapses or the message reception is interrupted
*/
Message receive(long timeout);
}

51
spring-eai-core/src/main/java/org/springframework/integration/MessageTarget.java
View File

@@ -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 <code>true</code> if the message is sent
* successfully, <code>false</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 <code>true</code> if the message is sent
* successfully, <code>false</false> if the specified
* timeout period elapses or the send is interrupted
*/
boolean send(Message message, long timeout);
}

35
spring-eai-core/src/main/java/org/springframework/integration/MessagingConfigurationException.java Normal file
View File

@@ -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);
}
}

41
spring-eai-core/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -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<String, MessageChannel> channelBeans = (Map<String, MessageChannel>) this.applicationContext
.getBeansOfType(MessageChannel.class);
private void registerChannels(ApplicationContext context) {
Map<String, MessageChannel> channelBeans =
(Map<String, MessageChannel>) context.getBeansOfType(MessageChannel.class);
for (Map.Entry<String, MessageChannel> 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<String, MessageEndpoint> endpointBeans = (Map<String, MessageEndpoint>) this.applicationContext
.getBeansOfType(MessageEndpoint.class);
private void registerEndpoints(ApplicationContext context) {
Map<String, MessageEndpoint> endpointBeans =
(Map<String, MessageEndpoint>) context.getBeansOfType(MessageEndpoint.class);
for (Map.Entry<String, MessageEndpoint> 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<String, Subscription> subscriptionBeans =
(Map<String, Subscription>) 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) {

57
spring-eai-core/src/main/java/org/springframework/integration/bus/Subscription.java Normal file
View File

@@ -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;
}
}

53
spring-eai-core/src/main/java/org/springframework/integration/channel/MessageChannel.java
View File

@@ -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 <code>true</code> if the message is sent
* successfully, <code>false</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 <code>true</code> if the message is sent
* successfully, <code>false</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
* <code>null</code> 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
* <code>null</code> if the specified timeout period
* elapses or the message reception is interrupted
*/
Message receive(long timeout);
}

162
spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractChannelAdapter.java Normal file
View File

@@ -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<Boolean> result = executor.submit(new Callable<Boolean>() {
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<Message> result = executor.submit(new Callable<Message>() {
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;
}

36
spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractInboundChannelAdapter.java Normal file
View File

@@ -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;
}

38
spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractOutboundChannelAdapter.java Normal file
View File

@@ -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;
}

66
spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingMessageSource.java
View File

@@ -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();
}

68
spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractPrefetchingMessageSource.java
View File

@@ -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<Object> queue = new LinkedBlockingQueue<Object>();
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();
}

28
spring-eai-core/src/main/java/org/springframework/integration/endpoint/ArgumentListPreparer.java Normal file
View File

@@ -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);
}

75
spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java
View File

@@ -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 '<i>replyChannelName</i>'.
* 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);
}
}

71
spring-eai-core/src/main/java/org/springframework/integration/endpoint/InboundMethodInvokingChannelAdapter.java Normal file
View File

@@ -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<T> extends AbstractInboundChannelAdapter {
private T object;
private String method;
private SimpleMethodInvoker<T> 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<T>(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.");
}
}
}
}

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

@@ -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);

66
spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageHandlerAdapter.java Normal file
View File

@@ -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<T> implements MessageHandler, InitializingBean {
private T object;
private String method;
private MessageMapper mapper;
private SimpleMethodInvoker<T> 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<T>(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;
}
}

33
spring-eai-core/src/main/java/org/springframework/integration/endpoint/MethodValidator.java Normal file
View File

@@ -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;
}

75
spring-eai-core/src/main/java/org/springframework/integration/endpoint/OutboundMethodInvokingChannelAdapter.java Normal file
View File

@@ -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<T> extends AbstractOutboundChannelAdapter {
private T object;
private String method;
private SimpleMethodInvoker<T> 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<T>(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;
}
}

73
spring-eai-core/src/main/java/org/springframework/integration/endpoint/SimpleMethodInvoker.java Normal file
View File

@@ -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<T> {
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);
}
}
}

26
spring-eai-core/src/test/java/org/springframework/integration/bus/EventDrivenConsumerTests.java
View File

@@ -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());
}

13
spring-eai-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java
View File

@@ -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)));

14
spring-eai-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java
View File

@@ -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);
}
}

12
spring-eai-core/src/test/java/org/springframework/integration/MessageBusTests.java → spring-eai-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java
View File

@@ -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());
}

4
spring-eai-core/src/test/java/org/springframework/integration/messageBusTests.xml → spring-eai-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml
View File

@@ -11,8 +11,8 @@
<bean id="targetChannel" class="org.springframework.integration.channel.PointToPointChannel"/>
<bean id="endpoint" class="org.springframework.integration.endpoint.GenericMessageEndpoint">
<property name="source" ref="sourceChannel"/>
<property name="target" ref="targetChannel"/>
<property name="inputChannelName" value="sourceChannel"/>
<property name="defaultOutputChannelName" value="targetChannel"/>
</bean>
</beans>

21
spring-eai-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java
View File

@@ -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");