renamed from spring-eai-core

This commit is contained in:
Mark Fisher
2007-12-15 18:25:49 +00:00
parent 2290442e27
commit c8259e436c
121 changed files with 0 additions and 0 deletions

View File

@@ -0,0 +1 @@
http\://www.springframework.org/schema/integration=org.springframework.integration.config.IntegrationNamespaceHandler

View File

@@ -0,0 +1 @@
http\://www.springframework.org/schema/integration/spring-integration-1.0.xsd=org/springframework/integration/config/spring-integration-1.0.xsd

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 error during message delivery.
*
* @author Mark Fisher
*/
@SuppressWarnings("serial")
public class MessageDeliveryException extends MessagingException {
public MessageDeliveryException(String message) {
super(message);
}
public MessageDeliveryException(String message, Throwable cause) {
super(message, cause);
}
}

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 error during message handling.
*
* @author Mark Fisher
*/
@SuppressWarnings("serial")
public class MessageHandlingException extends MessagingException {
public MessageHandlingException(String message) {
super(message);
}
public MessageHandlingException(String message, Throwable cause) {
super(message, cause);
}
}

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

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;
/**
* The base exception for any failures within the messaging system.
*
* @author Mark Fisher
*/
@SuppressWarnings("serial")
public class MessagingException extends RuntimeException {
public MessagingException(String message) {
super(message);
}
public MessagingException(String message, Throwable cause) {
super(message, cause);
}
}

View File

@@ -0,0 +1,31 @@
/*
* 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;
/**
* An unchecked wrapper exception for InterruptedExceptions.
*
* @author Mark Fisher
*/
@SuppressWarnings("serial")
public class SystemInterruptedException extends MessagingException {
public SystemInterruptedException(String message, Throwable cause) {
super(message, cause);
}
}

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.aop;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.aop.support.AopUtils;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.channel.ChannelMapping;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.util.Assert;
/**
* {@link MessagePublishingInterceptor} that resolves the channel from the
* publisher annotation of the invoked method.
*
* @author Mark Fisher
*/
public class AnnotationAwareMessagePublishingInterceptor extends MessagePublishingInterceptor {
private Class<? extends Annotation> publisherAnnotationType;
private String channelAttributeName;
private ChannelMapping channelMapping;
public AnnotationAwareMessagePublishingInterceptor(Class<? extends Annotation> publisherAnnotationType,
String channelAttributeName, ChannelMapping channelMapping) {
Assert.notNull(publisherAnnotationType, "publisherAnnotationType must not be null");
Assert.notNull(channelAttributeName, "channelAttributeName must not be null");
Assert.notNull(channelMapping, "channelMapping must not be null");
this.publisherAnnotationType = publisherAnnotationType;
this.channelAttributeName = channelAttributeName;
this.channelMapping = channelMapping;
}
@Override
protected MessageChannel resolveChannel(MethodInvocation invocation) {
Class<?> targetClass = AopUtils.getTargetClass(invocation.getThis());
Method method = AopUtils.getMostSpecificMethod(invocation.getMethod(), targetClass);
Annotation annotation = AnnotationUtils.getAnnotation(method, this.publisherAnnotationType);
if (annotation != null) {
String channelName = (String) AnnotationUtils.getValue(annotation, this.channelAttributeName);
if (channelName != null) {
MessageChannel channel = this.channelMapping.getChannel(channelName);
if (channel != null) {
return channel;
}
}
}
return super.resolveChannel(invocation);
}
}

View File

@@ -0,0 +1,86 @@
/*
* 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.aop;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.MessageMapper;
import org.springframework.integration.message.SimplePayloadMessageMapper;
import org.springframework.util.Assert;
/**
* Interceptor that publishes a target method's return value to a channel.
*
* @author Mark Fisher
*/
public class MessagePublishingInterceptor implements MethodInterceptor {
protected Log logger = LogFactory.getLog(getClass());
private MessageMapper mapper = new SimplePayloadMessageMapper();
private MessageChannel defaultChannel;
public void setDefaultChannel(MessageChannel defaultChannel) {
this.defaultChannel = defaultChannel;
}
/**
* Specify the {@link MessageMapper} to use when creating a message from the
* return value Object. The default is a {@link SimplePayloadMessageMapper}.
*
* @param mapper the mapper to use
*/
public void setMessageMapper(MessageMapper mapper) {
Assert.notNull(mapper, "mapper must not be null");
this.mapper = mapper;
}
/**
* Invoke the target method and publish its return value.
*/
public Object invoke(MethodInvocation invocation) throws Throwable {
Object retval = invocation.proceed();
if (retval != null) {
MessageChannel channel = this.resolveChannel(invocation);
if (channel == null) {
if (logger.isWarnEnabled()) {
logger.warn("unable to resolve channel for intercepted method '" +
invocation.getMethod().getName() + "'");
}
}
else {
channel.send(mapper.toMessage(retval));
}
}
return retval;
}
/**
* Subclasses may override this method to provide custom behavior.
*/
protected MessageChannel resolveChannel(MethodInvocation invocation) {
return this.defaultChannel;
}
}

View File

@@ -0,0 +1,40 @@
/*
* 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.aop;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
* Indicates that the method's return value should be published to the specified
* channel. The value will only be published if non-null.
*
* @author Mark Fisher
*/
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface Publisher {
String channel();
}

View File

@@ -0,0 +1,68 @@
/*
* Copyright 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aop;
import java.lang.annotation.Annotation;
import org.aopalliance.aop.Advice;
import org.springframework.aop.Pointcut;
import org.springframework.aop.support.AbstractPointcutAdvisor;
import org.springframework.aop.support.annotation.AnnotationMatchingPointcut;
import org.springframework.integration.channel.ChannelMapping;
import org.springframework.util.Assert;
/**
* Advisor whose pointcut matches a method annotation and whose advice will
* publish a message to the channel provided by that annotation.
*
* @author Mark Fisher
* @see Publisher
* @see AnnotationAwareMessagePublishingInterceptor
*/
@SuppressWarnings("serial")
public class PublisherAnnotationAdvisor extends AbstractPointcutAdvisor {
private AnnotationAwareMessagePublishingInterceptor advice;
private AnnotationMatchingPointcut pointcut;
public PublisherAnnotationAdvisor(ChannelMapping channelMapping) {
this(Publisher.class, "channel", channelMapping);
}
public PublisherAnnotationAdvisor(Class<? extends Annotation> publisherAnnotationType, String channelNameAttribute,
ChannelMapping channelMapping) {
Assert.notNull(publisherAnnotationType, "publisherAnnotationType must not be null");
Assert.notNull(channelNameAttribute, "channelNameAttribute must not be null");
Assert.notNull(channelMapping, "channelMapping must not be null");
this.pointcut = AnnotationMatchingPointcut.forMethodAnnotation(publisherAnnotationType);
this.advice = new AnnotationAwareMessagePublishingInterceptor(publisherAnnotationType, channelNameAttribute,
channelMapping);
}
public Pointcut getPointcut() {
return this.pointcut;
}
public Advice getAdvice() {
return this.advice;
}
}

View File

@@ -0,0 +1,154 @@
/*
* 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;
import java.util.concurrent.TimeUnit;
/**
* A container for Message consumer configuration metadata.
*
* @author Mark Fisher
*/
public class ConsumerPolicy {
private static final int DEFAULT_CONCURRENCY = 1;
private static final int DEFAULT_MAX_CONCURRENCY = 10;
private static final int DEFAULT_MAX_MESSAGES_PER_TASK = 10;
private static final int DEFAULT_REJECTION_LIMIT = 10;
private static final int DEFAULT_REJECTION_LIMIT_WAIT = 1000;
private static final long DEFAULT_RECEIVE_TIMEOUT = 1000;
private int concurrency = DEFAULT_CONCURRENCY;
private int maxConcurrency = DEFAULT_MAX_CONCURRENCY;
private int maxMessagesPerTask = DEFAULT_MAX_MESSAGES_PER_TASK;
private int rejectionLimit = DEFAULT_REJECTION_LIMIT;
private int rejectionLimitWait = DEFAULT_REJECTION_LIMIT_WAIT;
private int initialDelay = 0;
private int period = 5;
private TimeUnit timeUnit = TimeUnit.MILLISECONDS;
private boolean fixedRate = false;
private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
public int getInitialDelay() {
return this.initialDelay;
}
public void setInitialDelay(int initialDelay) {
this.initialDelay = initialDelay;
}
public int getPeriod() {
return this.period;
}
public void setPeriod(int period) {
this.period = period;
}
public TimeUnit getTimeUnit() {
return this.timeUnit;
}
public void setTimeUnit(TimeUnit timeUnit) {
this.timeUnit = timeUnit;
}
public boolean isFixedRate() {
return this.fixedRate;
}
public void setFixedRate(boolean fixedRate) {
this.fixedRate = fixedRate;
}
public int getConcurrency() {
return this.concurrency;
}
public void setConcurrency(int concurrency) {
if (concurrency < 1) {
throw new IllegalArgumentException("'concurrency' value must be at least 1");
}
this.concurrency = concurrency;
}
public int getMaxConcurrency() {
return this.maxConcurrency;
}
public void setMaxConcurrency(int maxConcurrency) {
if (maxConcurrency < 1) {
throw new IllegalArgumentException("'maxConcurrency' value must be at least 1");
}
this.maxConcurrency = maxConcurrency;
}
public int getMaxMessagesPerTask() {
return this.maxMessagesPerTask;
}
public void setMaxMessagesPerTask(int maxMessagesPerTask) {
if (maxMessagesPerTask == 0) {
throw new IllegalArgumentException("'maxMessagesPerTask' must not be 0");
}
this.maxMessagesPerTask = maxMessagesPerTask;
}
public int getRejectionLimit() {
return this.rejectionLimit;
}
public void setRejectionLimit(int rejectionLimit) {
if (rejectionLimit < 1) {
throw new IllegalArgumentException("'idleTaskExecutionLimit' must be at least 1");
}
this.rejectionLimit = rejectionLimit;
}
public int getRejectionLimitWait() {
return this.rejectionLimitWait;
}
public void setRejectionLimitWait(int rejectionLimitWait) {
this.rejectionLimitWait = rejectionLimitWait;
}
public long getReceiveTimeout() {
return this.receiveTimeout;
}
public void setReceiveTimeout(long receiveTimeout) {
this.receiveTimeout = receiveTimeout;
}
}

View File

@@ -0,0 +1,90 @@
/*
* 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;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.bus.MessageBus.EndpointTask;
/**
* A subclass of {@link ThreadPoolExecutor} with configurable error thresholds.
*
* @author Mark Fisher
*/
public class EndpointExecutor extends ThreadPoolExecutor {
private Log logger = LogFactory.getLog(this.getClass());
private int successiveErrorCount;
private int successiveErrorThreshold = -1;
private int totalErrorCount;
private int totalErrorThreshold = -1;
public EndpointExecutor(int corePoolSize, int maximumPoolSize) {
super(corePoolSize, maximumPoolSize, 0, TimeUnit.MILLISECONDS, new SynchronousQueue<Runnable>());
}
/**
* Set the maximum number of errors allowed in <em>successive</em>
* endpoint executions. If this threshold is ever exceeded, the executor
* will shutdown.
*/
public void setSuccessiveErrorThreshold(int successiveErrorThreshold) {
this.successiveErrorThreshold = successiveErrorThreshold;
}
/**
* Set the maximum number of <em>total</em> errors allowed in endpoint
* executions. If this threshold is ever exceeded, the executor will
* shutdown.
*/
public void setTotalErrorThreshold(int totalErrorThreshold) {
this.totalErrorThreshold = totalErrorThreshold;
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
EndpointTask task = (EndpointTask) r;
if (task.getError() != null) {
if (logger.isWarnEnabled()) {
logger.warn("Exception occurred in endpoint execution: " + task.getError());
}
this.successiveErrorCount++;
this.totalErrorCount++;
if ((successiveErrorThreshold >= 0 && successiveErrorCount > successiveErrorThreshold)
|| (totalErrorThreshold >= 0 && totalErrorCount > totalErrorThreshold)) {
if (logger.isInfoEnabled()) {
logger.info("error threshold exceeded, shutting down now");
}
this.shutdownNow();
}
}
else {
successiveErrorCount = 0;
}
}
}

View File

@@ -0,0 +1,355 @@
/*
* 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;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.Lifecycle;
import org.springframework.integration.MessagingException;
import org.springframework.integration.channel.ChannelMapping;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PointToPointChannel;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.message.Message;
import org.springframework.util.Assert;
/**
* The messaging bus. Serves as a registry for channels and endpoints, manages their lifecycle,
* and activates subscriptions.
*
* @author Mark Fisher
*/
public class MessageBus implements ChannelMapping, ApplicationContextAware, Lifecycle {
private Log logger = LogFactory.getLog(this.getClass());
private Map<String, MessageChannel> channels = new ConcurrentHashMap<String, MessageChannel>();
private Map<String, MessageEndpoint> endpoints = new ConcurrentHashMap<String, MessageEndpoint>();
private List<DispatcherTask> dispatcherTasks = new CopyOnWriteArrayList<DispatcherTask>();
private Map<MessageEndpoint, EndpointExecutor> endpointExecutors = new ConcurrentHashMap<MessageEndpoint, EndpointExecutor>();
private ScheduledThreadPoolExecutor dispatcherExecutor;
private MessageChannel invalidMessageChannel;
private boolean autoCreateChannels;
private boolean running;
private Object lifecycleMonitor = new Object();
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
Assert.notNull(applicationContext, "applicationContext must not be null");
this.registerChannels(applicationContext);
this.registerEndpoints(applicationContext);
this.activateSubscriptions(applicationContext);
}
public void setAutoCreateChannels(boolean autoCreateChannels) {
this.autoCreateChannels = autoCreateChannels;
}
public void setInvalidMessageChannel(MessageChannel invalidMessageChannel) {
this.invalidMessageChannel = invalidMessageChannel;
}
@SuppressWarnings("unchecked")
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()) {
logger.info("registered channel '" + entry.getKey() + "'");
}
}
}
@SuppressWarnings("unchecked")
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()) {
logger.info("registered endpoint '" + entry.getKey() + "'");
}
}
}
@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);
}
public MessageChannel getChannel(String channelName) {
return this.channels.get(channelName);
}
public MessageChannel getInvalidMessageChannel() {
return this.invalidMessageChannel;
}
public void registerChannel(String name, MessageChannel channel) {
this.channels.put(name, channel);
}
public void registerEndpoint(String name, MessageEndpoint endpoint) {
this.endpoints.put(name, endpoint);
endpoint.setChannelMapping(this);
if (endpoint.getInputChannelName() != null && endpoint.getConsumerPolicy() != null) {
Subscription subscription = new Subscription();
subscription.setChannel(endpoint.getInputChannelName());
subscription.setEndpoint(name);
subscription.setPolicy(endpoint.getConsumerPolicy());
this.activateSubscription(subscription);
}
}
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) {
throw new MessagingException("Cannot activate subscription, unknown channel '" + channelName +
"'. Consider enabling the 'autoCreateChannels' option for the message bus.");
}
this.registerChannel(channelName, new PointToPointChannel());
if (this.logger.isInfoEnabled()) {
logger.info("created channel '" + channelName + "'");
}
}
MessageEndpoint endpoint = this.endpoints.get(endpointName);
if (endpoint == null) {
throw new MessagingException("Cannot activate subscription, unknown endpoint '" + endpointName + "'");
}
EndpointExecutor endpointExecutor = new EndpointExecutor(policy.getConcurrency(), policy.getMaxConcurrency());
endpointExecutors.put(endpoint, endpointExecutor);
DispatcherTask dispatcherTask = new DispatcherTask(channel, endpoint, policy);
this.dispatcherTasks.add(dispatcherTask);
if (this.isRunning()) {
scheduleDispatcherTask(dispatcherTask);
if (this.logger.isInfoEnabled()) {
logger.info("scheduled dispatcher task: channel='" +
channelName + "' endpoint='" + endpointName + "'");
}
}
}
public int getActiveCountForEndpoint(String endpointName) {
MessageEndpoint endpoint = this.endpoints.get(endpointName);
if (endpoint != null) {
EndpointExecutor executor = this.endpointExecutors.get(endpoint);
if (executor != null) {
return executor.getActiveCount();
}
}
return 0;
}
private void scheduleDispatcherTask(DispatcherTask task) {
ConsumerPolicy policy = task.getPolicy();
if (policy.getPeriod() <= 0) {
if (policy.getReceiveTimeout() <= 0) {
if (logger.isWarnEnabled()) {
logger.warn("Scheduling a repeating task with no receive timeout is not recommended! " +
"Consider providing a positive value for either 'period' or 'receiveTimeout'");
}
}
dispatcherExecutor.schedule(new RepeatingDispatcherTask(task), policy.getInitialDelay(), policy.getTimeUnit());
}
else if (policy.isFixedRate()) {
dispatcherExecutor.scheduleAtFixedRate(task, policy.getInitialDelay(), policy.getPeriod(), policy.getTimeUnit());
}
else {
dispatcherExecutor.scheduleWithFixedDelay(task, policy.getInitialDelay(), policy.getPeriod(), policy.getTimeUnit());
}
}
public boolean isRunning() {
synchronized (this.lifecycleMonitor) {
return this.running;
}
}
public void start() {
if (this.dispatcherExecutor == null) {
this.initialize();
}
synchronized (this.lifecycleMonitor) {
if (!this.isRunning()) {
this.running = true;
for (DispatcherTask task : this.dispatcherTasks) {
scheduleDispatcherTask(task);
}
}
}
}
public void stop() {
synchronized (this.lifecycleMonitor) {
if (this.isRunning()) {
this.running = false;
this.dispatcherExecutor.shutdownNow();
}
}
}
private class DispatcherTask implements Runnable {
private MessageChannel channel;
private MessageEndpoint endpoint;
private ConsumerPolicy policy;
public DispatcherTask(MessageChannel channel, MessageEndpoint endpoint, ConsumerPolicy policy) {
this.channel = channel;
this.endpoint = endpoint;
this.policy = policy;
}
public MessageChannel getChannel() {
return this.channel;
}
public MessageEndpoint getEndpoint() {
return this.endpoint;
}
public ConsumerPolicy getPolicy() {
return this.policy;
}
public void run() {
EndpointExecutor executor = endpointExecutors.get(this.endpoint);
if (executor == null || executor.isShutdown()) {
if (logger.isWarnEnabled()) {
logger.warn("dispatcher shutting down, endpoint executor is not active");
}
return;
}
for (int i = 0; i < policy.getMaxMessagesPerTask(); i++) {
Message message = channel.receive(this.policy.getReceiveTimeout());
if (message == null) {
return;
}
else {
boolean taskSubmitted = false;
int attempts = 0;
while (!taskSubmitted) {
try {
executor.execute(new EndpointTask(this.endpoint, message));
taskSubmitted = true;
}
catch (RejectedExecutionException rex) {
attempts++;
if (attempts == policy.getRejectionLimit()) {
attempts = 0;
if (logger.isDebugEnabled()) {
logger.debug("reached rejected execution limit");
}
try {
Thread.sleep(policy.getRejectionLimitWait());
}
catch (InterruptedException iex) {
Thread.currentThread().interrupt();
}
}
}
}
}
}
}
}
private class RepeatingDispatcherTask implements Runnable {
private DispatcherTask task;
RepeatingDispatcherTask(DispatcherTask task) {
this.task = task;
}
public void run() {
task.run();
dispatcherExecutor.execute(new RepeatingDispatcherTask(task));
}
}
public static class EndpointTask implements Runnable {
private MessageEndpoint endpoint;
private Message message;
private Throwable error;
EndpointTask(MessageEndpoint endpoint, Message message) {
this.endpoint = endpoint;
this.message = message;
}
public Throwable getError() {
return this.error;
}
public void run() {
try {
this.endpoint.messageReceived(this.message);
}
catch (Throwable t) {
this.error = t;
}
}
}
}

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

View File

@@ -0,0 +1,30 @@
/*
* 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.channel;
/**
* A strategy interface for resolution of message channels by name.
*
* @author Mark Fisher
*/
public interface ChannelMapping {
MessageChannel getChannel(String channelName);
MessageChannel getInvalidMessageChannel();
}

View File

@@ -0,0 +1,34 @@
/*
* 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.channel;
/**
* Enumeration of the different types of message consumer.
*
* @author Mark Fisher
*/
public enum ConsumerType {
EVENT_DRIVEN,
FIXED_RATE,
FIXED_DELAY,
SCHEDULED
}

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.channel;
import org.springframework.integration.message.Message;
/**
* Base channel interface defining common behavior for message reception and sending.
*
* @author Mark Fisher
*/
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);
}

View File

@@ -0,0 +1,189 @@
/*
* 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.channel;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageSelector;
/**
* Simple implementation of a point-to-point message channel. Each Messages is
* placed in a queue whose capacity may be specified upon construction. If no
* capacity is specified, the {@link #DEFAULT_CAPACITY} will be used.
*
* @author Mark Fisher
*/
public class PointToPointChannel implements MessageChannel {
private static final int DEFAULT_CAPACITY = 25;
private BlockingQueue<Message> queue;
/**
* Create a channel with the specified queue capacity.
*/
public PointToPointChannel(int capacity) {
queue = new LinkedBlockingQueue<Message>(capacity);
}
/**
* Create a channel with the default queue capacity.
*/
public PointToPointChannel() {
this(DEFAULT_CAPACITY);
}
/**
* Send a message on this channel. If the queue is full, this method will
* block until either space becomes available or the sending thread is
* interrupted.
*
* @param message the Message to send
*
* @return <code>true</code> if the message is sent successfully or
* <code>false</code> if the sending thread is interrupted.
*/
public boolean send(Message message) {
try {
queue.put(message);
return true;
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
return false;
}
}
/**
* Send a message on this channel. If the queue is full, this method will
* block until either the timeout occurs or the sending thread is
* interrupted. If the specified timeout is less than 1, the method will
* return immediately.
*
* @param message the Message to send
* @param timeout the timeout in milliseconds
*
* @return <code>true</code> if the message is sent successfully,
* <code>false</code> if the message cannot be sent within the allotted
* time or the sending thread is interrupted.
*/
public boolean send(Message message, long timeout) {
try {
if (timeout > 0) {
return this.queue.offer(message, timeout, TimeUnit.MILLISECONDS);
}
return this.queue.offer(message);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
return false;
}
}
/**
* Receive the message at the head of the queue. If the queue is empty, this
* method will block.
*
* @return the Message at the head of the queue or <code>null</code> if
* the receiving thread is interrupted.
*/
public Message receive() {
try {
return queue.take();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
return null;
}
}
/**
* Receive the message at the head of the queue. If the queue is empty, this
* method will block until the allotted timeout elapses. If the specified
* timeout is less than 1, the method will return immediately.
*
* @param timeout the timeout in milliseconds
*
* @return the message at the head of the queue or <code>null</code> in
* case no message is available within the allotted time or the receiving
* thread is interrupted.
*/
public Message receive(long timeout) {
try {
if (timeout > 0) {
return queue.poll(timeout, TimeUnit.MILLISECONDS);
}
return queue.poll();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
return null;
}
}
/**
* Receive the first message that is accepted by the specified selector
* starting from the head of the queue. If the queue is empty, this method
* will block until the allotted timeout elapses. If the specified timeout
* is less than 1, the method will return immediately.
*
* @param selector the selector to use
* @param timeout the timeout in milliseconds
*
* @return the first accepted message or <code>null</code> in case the
* selector does not accept any message within the allotted time or the
* receiving thread is interrupted.
*/
public Message receive(MessageSelector selector, long timeout) {
long start = System.currentTimeMillis();
while (timeout <= 0 || System.currentTimeMillis() - start < timeout) {
Object[] elements = this.queue.toArray();
for (int i = (elements.length - 1); i >= 0; i--) {
Message m = (Message) elements[i];
if (selector.accept(m) && this.queue.remove(m)) {
return m;
}
}
if (timeout == 0) {
return null;
}
}
return null;
}
/**
* Receive the first message that is accepted by the specified selector
* starting from the head of the queue. If the queue is empty, this method
* will block.
*
* @param selector the selector to use
*
* @return the first accepted message or <code>null</code> in case the
* selector does not accept any message or the receiving thread is
* interrupted.
*/
public Message receive(MessageSelector selector) {
return this.receive(selector, -1);
}
}

View File

@@ -0,0 +1,81 @@
/*
* Copyright 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.BeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.endpoint.annotation.MessageEndpointAnnotationPostProcessor;
/**
* Parser for the <em>annotation-driven</em> element of the integration
* namespace. Registers the annotation-driven post-processors.
*
* @author Mark Fisher
*/
public class AnnotationDrivenParser implements BeanDefinitionParser {
private static final String PUBLISHER_ANNOTATION_POST_PROCESSOR_BEAN_NAME =
"internal.PublisherAnnotationPostProcessor";
private static final String SUBSCRIBER_ANNOTATION_POST_PROCESSOR_BEAN_NAME =
"internal.SubscriberAnnotationPostProcessor";
private static final String MESSAGE_ENDPOINT_ANNOTATION_POST_PROCESSOR_BEAN_NAME =
"internal.MessageEndpointAnnotationPostProcessor";
public BeanDefinition parse(Element element, ParserContext parserContext) {
this.createPublisherPostProcessor(parserContext);
this.createSubscriberPostProcessor(parserContext);
this.createMessageEndpointPostProcessor(parserContext);
return null;
}
private void createPublisherPostProcessor(ParserContext parserContext) {
BeanDefinition bd = new RootBeanDefinition(PublisherAnnotationPostProcessor.class);
bd.getPropertyValues().addPropertyValue("channelMapping",
new RuntimeBeanReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME));
BeanComponentDefinition bcd = new BeanComponentDefinition(
bd, PUBLISHER_ANNOTATION_POST_PROCESSOR_BEAN_NAME);
parserContext.registerBeanComponent(bcd);
}
private void createSubscriberPostProcessor(ParserContext parserContext) {
BeanDefinition bd = new RootBeanDefinition(SubscriberAnnotationPostProcessor.class);
bd.getPropertyValues().addPropertyValue("messageBus",
new RuntimeBeanReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME));
BeanComponentDefinition bcd = new BeanComponentDefinition(
bd, SUBSCRIBER_ANNOTATION_POST_PROCESSOR_BEAN_NAME);
parserContext.registerBeanComponent(bcd);
}
private void createMessageEndpointPostProcessor(ParserContext parserContext) {
BeanDefinition bd = new RootBeanDefinition(MessageEndpointAnnotationPostProcessor.class);
bd.getPropertyValues().addPropertyValue("messageBus",
new RuntimeBeanReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME));
BeanComponentDefinition bcd = new BeanComponentDefinition(
bd, MESSAGE_ENDPOINT_ANNOTATION_POST_PROCESSOR_BEAN_NAME);
parserContext.registerBeanComponent(bcd);
}
}

View File

@@ -0,0 +1,76 @@
/*
* Copyright 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.BeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.endpoint.InboundMethodInvokingChannelAdapter;
import org.springframework.integration.endpoint.OutboundMethodInvokingChannelAdapter;
import org.springframework.util.StringUtils;
/**
* Parser for inbound and outbound channel adapters.
*
* @author Mark Fisher
*/
public class ChannelAdapterParser implements BeanDefinitionParser {
private static final String ID_ATTRIBUTE = "id";
private static final String REF_ATTRIBUTE = "ref";
private static final String METHOD_ATTRIBUTE = "method";
private final boolean isInbound;
public ChannelAdapterParser(boolean isInbound) {
this.isInbound = isInbound;
}
public BeanDefinition parse(Element element, ParserContext parserContext) {
RootBeanDefinition adapterDef = null;
if (this.isInbound) {
adapterDef = new RootBeanDefinition(InboundMethodInvokingChannelAdapter.class);
}
else {
adapterDef = new RootBeanDefinition(OutboundMethodInvokingChannelAdapter.class);
}
adapterDef.setSource(parserContext.extractSource(element));
String ref = element.getAttribute(REF_ATTRIBUTE);
String method = element.getAttribute(METHOD_ATTRIBUTE);
if (!StringUtils.hasText(ref) || !StringUtils.hasText(method)) {
}
adapterDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(ref));
adapterDef.getPropertyValues().addPropertyValue("method", method);
String beanName = element.getAttribute(ID_ATTRIBUTE);
if (!StringUtils.hasText(beanName)) {
beanName = parserContext.getReaderContext().generateBeanName(adapterDef);
}
parserContext.registerBeanComponent(new BeanComponentDefinition(adapterDef, beanName));
return adapterDef;
}
}

View File

@@ -0,0 +1,56 @@
/*
* Copyright 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.BeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.channel.PointToPointChannel;
import org.springframework.util.StringUtils;
/**
* Parser for the <em>channel</em> element of the integration namespace.
*
* @author Mark Fisher
*/
public class ChannelParser implements BeanDefinitionParser {
private static final String ID_ATTRIBUTE = "id";
private static final String CAPACITY_ATTRIBUTE = "capacity";
public BeanDefinition parse(Element element, ParserContext parserContext) {
RootBeanDefinition channelDef = new RootBeanDefinition(PointToPointChannel.class);
channelDef.setSource(parserContext.extractSource(element));
String capacity = element.getAttribute(CAPACITY_ATTRIBUTE);
if (StringUtils.hasText(capacity)) {
channelDef.getConstructorArgumentValues().addGenericArgumentValue(Integer.parseInt(capacity));
}
String beanName = element.getAttribute(ID_ATTRIBUTE);
if (!StringUtils.hasText(beanName)) {
beanName = parserContext.getReaderContext().generateBeanName(channelDef);
}
parserContext.registerBeanComponent(new BeanComponentDefinition(channelDef, beanName));
return channelDef;
}
}

View File

@@ -0,0 +1,85 @@
/*
* Copyright 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.config;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.BeanNameGenerator;
import org.springframework.beans.factory.support.DefaultBeanNameGenerator;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.integration.endpoint.GenericMessageEndpoint;
import org.springframework.integration.endpoint.InboundMethodInvokingChannelAdapter;
import org.springframework.integration.endpoint.MessageHandlerAdapter;
import org.springframework.integration.endpoint.OutboundMethodInvokingChannelAdapter;
import org.springframework.util.Assert;
/**
* Factory for creating integration component bean definitions.
*
* @author Mark Fisher
*/
public class ComponentConfigurer {
private BeanDefinitionRegistry registry;
private BeanNameGenerator beanNameGenerator;
public ComponentConfigurer(BeanDefinitionRegistry registry, BeanNameGenerator beanNameGenerator) {
Assert.notNull(registry, "registry must not be null");
this.registry = registry;
this.beanNameGenerator = (beanNameGenerator != null ? beanNameGenerator : new DefaultBeanNameGenerator());
}
public String serviceActivator(String inputChannel, String outputChannel, String objectRef, String method) {
RootBeanDefinition endpointDef = new RootBeanDefinition(GenericMessageEndpoint.class);
RootBeanDefinition adapterDef = new RootBeanDefinition(MessageHandlerAdapter.class);
adapterDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(objectRef));
adapterDef.getPropertyValues().addPropertyValue("method", method);
String adapterName = beanNameGenerator.generateBeanName(adapterDef, this.registry);
this.registry.registerBeanDefinition(adapterName, adapterDef);
endpointDef.getPropertyValues().addPropertyValue("handler", new RuntimeBeanReference(adapterName));
if (inputChannel != null) {
endpointDef.getPropertyValues().addPropertyValue("inputChannelName", inputChannel);
}
if (outputChannel != null) {
endpointDef.getPropertyValues().addPropertyValue("defaultOutputChannelName", outputChannel);
}
String endpointName = beanNameGenerator.generateBeanName(endpointDef, this.registry);
this.registry.registerBeanDefinition(endpointName, endpointDef);
return endpointName;
}
public String inboundChannelAdapter(String objectRef, String method) {
RootBeanDefinition bd = new RootBeanDefinition(InboundMethodInvokingChannelAdapter.class);
bd.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(objectRef));
bd.getPropertyValues().addPropertyValue("method", method);
String beanName = this.beanNameGenerator.generateBeanName(bd, this.registry);
this.registry.registerBeanDefinition(beanName, bd);
return beanName;
}
public String outboundChannelAdapter(String objectRef, String method) {
RootBeanDefinition bd = new RootBeanDefinition(OutboundMethodInvokingChannelAdapter.class);
bd.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(objectRef));
bd.getPropertyValues().addPropertyValue("method", method);
String beanName = this.beanNameGenerator.generateBeanName(bd, this.registry);
this.registry.registerBeanDefinition(beanName, bd);
return beanName;
}
}

View File

@@ -0,0 +1,127 @@
/*
* Copyright 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.config;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.BeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.bus.ConsumerPolicy;
import org.springframework.integration.endpoint.GenericMessageEndpoint;
import org.springframework.integration.endpoint.MessageHandlerAdapter;
import org.springframework.util.StringUtils;
/**
* Parser for the <em>endpoint</em> element of the integration namespace.
*
* @author Mark Fisher
*/
public class EndpointParser implements BeanDefinitionParser {
private static final String ID_ATTRIBUTE = "id";
private static final String INPUT_CHANNEL_ATTRIBUTE = "input-channel";
private static final String INPUT_CHANNEL_PROPERTY = "inputChannelName";
private static final String DEFAULT_OUTPUT_CHANNEL_ATTRIBUTE = "default-output-channel";
private static final String DEFAULT_OUTPUT_CHANNEL_PROPERTY = "defaultOutputChannelName";
private static final String HANDLER_REF_ATTRIBUTE = "handler-ref";
private static final String HANDLER_METHOD_ATTRIBUTE = "handler-method";
private static final String HANDLER_PROPERTY = "handler";
private static final String OBJECT_PROPERTY = "object";
private static final String METHOD_PROPERTY = "method";
private static final String PERIOD_ATTRIBUTE = "period";
private static final String PERIOD_PROPERTY = "period";
private static final String CONSUMER_ELEMENT = "consumer";
private static final String CONSUMER_POLICY_PROPERTY = "consumerPolicy";
public BeanDefinition parse(Element element, ParserContext parserContext) {
RootBeanDefinition endpointDef = new RootBeanDefinition(GenericMessageEndpoint.class);
endpointDef.setSource(parserContext.extractSource(element));
String inputChannel = element.getAttribute(INPUT_CHANNEL_ATTRIBUTE);
if (StringUtils.hasText(inputChannel)) {
endpointDef.getPropertyValues().addPropertyValue(INPUT_CHANNEL_PROPERTY, inputChannel);
}
String defaultOutputChannel = element.getAttribute(DEFAULT_OUTPUT_CHANNEL_ATTRIBUTE);
if (StringUtils.hasText(defaultOutputChannel)) {
endpointDef.getPropertyValues().addPropertyValue(DEFAULT_OUTPUT_CHANNEL_PROPERTY, defaultOutputChannel);
}
String handlerRef = element.getAttribute(HANDLER_REF_ATTRIBUTE);
if (StringUtils.hasText(handlerRef)) {
String handlerMethod = element.getAttribute(HANDLER_METHOD_ATTRIBUTE);
if (StringUtils.hasText(handlerMethod)) {
BeanDefinition handlerAdapterDef = new RootBeanDefinition(MessageHandlerAdapter.class);
handlerAdapterDef.getPropertyValues().addPropertyValue(OBJECT_PROPERTY, new RuntimeBeanReference(handlerRef));
handlerAdapterDef.getPropertyValues().addPropertyValue(METHOD_PROPERTY, handlerMethod);
String adapterBeanName = parserContext.getReaderContext().generateBeanName(handlerAdapterDef);
parserContext.registerBeanComponent(new BeanComponentDefinition(handlerAdapterDef, adapterBeanName));
endpointDef.getPropertyValues().addPropertyValue(HANDLER_PROPERTY, new RuntimeBeanReference(adapterBeanName));
}
else {
endpointDef.getPropertyValues().addPropertyValue(HANDLER_PROPERTY, new RuntimeBeanReference(handlerRef));
}
}
String beanName = element.getAttribute(ID_ATTRIBUTE);
if (!StringUtils.hasText(beanName)) {
beanName = parserContext.getReaderContext().generateBeanName(endpointDef);
}
NodeList childNodes = element.getChildNodes();
for (int i = 0; i < childNodes.getLength(); i++) {
Node child = childNodes.item(i);
if (child.getNodeType() == Node.ELEMENT_NODE) {
String localName = child.getLocalName();
if (CONSUMER_ELEMENT.equals(localName)) {
String consumerBeanName = parseConsumer((Element) child, parserContext);
endpointDef.getPropertyValues().addPropertyValue(
CONSUMER_POLICY_PROPERTY, new RuntimeBeanReference(consumerBeanName));
}
}
}
parserContext.registerBeanComponent(new BeanComponentDefinition(endpointDef, beanName));
return endpointDef;
}
private String parseConsumer(Element element, ParserContext parserContext) {
RootBeanDefinition consumerDef = new RootBeanDefinition(ConsumerPolicy.class);
String period = element.getAttribute(PERIOD_ATTRIBUTE);
if (StringUtils.hasText(period)) {
consumerDef.getPropertyValues().addPropertyValue(PERIOD_PROPERTY, Integer.parseInt(period));
}
String beanName = parserContext.getReaderContext().generateBeanName(consumerDef);
parserContext.registerBeanComponent(new BeanComponentDefinition(consumerDef, beanName));
return beanName;
}
}

View File

@@ -0,0 +1,37 @@
/*
* Copyright 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.config;
import org.springframework.beans.factory.xml.NamespaceHandlerSupport;
/**
* Namespace handler for the integration namespace.
*
* @author Mark Fisher
*/
public class IntegrationNamespaceHandler extends NamespaceHandlerSupport {
public void init() {
registerBeanDefinitionParser("message-bus", new MessageBusParser());
registerBeanDefinitionParser("annotation-driven", new AnnotationDrivenParser());
registerBeanDefinitionParser("channel", new ChannelParser());
registerBeanDefinitionParser("inbound-channel-adapter", new ChannelAdapterParser(true));
registerBeanDefinitionParser("outbound-channel-adapter", new ChannelAdapterParser(false));
registerBeanDefinitionParser("endpoint", new EndpointParser());
}
}

View File

@@ -0,0 +1,47 @@
/*
* Copyright 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.bus.MessageBus;
/**
* Parser for the <em>message-bus</em> element of the integration namespace.
*
* @author Mark Fisher
*/
public class MessageBusParser extends AbstractSingleBeanDefinitionParser {
public static final String MESSAGE_BUS_BEAN_NAME = "internal.messageBus";
@Override
protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
throws BeanDefinitionStoreException {
return MESSAGE_BUS_BEAN_NAME;
}
@Override
protected Class<?> getBeanClass(Element element) {
return MessageBus.class;
}
}

View File

@@ -0,0 +1,109 @@
/*
* Copyright 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.config;
import java.lang.annotation.Annotation;
import org.springframework.aop.Advisor;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.integration.aop.Publisher;
import org.springframework.integration.aop.PublisherAnnotationAdvisor;
import org.springframework.integration.channel.ChannelMapping;
import org.springframework.util.Assert;
/**
* A {@link BeanPostProcessor} that adds a message publishing interceptor when
* it discovers annotated methods.
*
* @author Mark Fisher
*/
public class PublisherAnnotationPostProcessor implements BeanPostProcessor, BeanClassLoaderAware {
private Class<? extends Annotation> publisherAnnotationType = Publisher.class;
private String channelNameAttribute = "channel";
private ChannelMapping channelMapping;
private Advisor advisor;
private ClassLoader beanClassLoader;
public void setBeanClassLoader(ClassLoader beanClassLoader) {
Assert.notNull(beanClassLoader, "beanClassLoader must not be null");
this.beanClassLoader = beanClassLoader;
}
public void setPublisherAnnotationType(Class<? extends Annotation> publisherAnnotationType) {
Assert.notNull(publisherAnnotationType, "publisherAnnotationType must not be null");
this.publisherAnnotationType = publisherAnnotationType;
}
public void setChannelNameAttribute(String channelNameAttribute) {
Assert.notNull(channelNameAttribute, "channelNameAttribute must not be null");
this.channelNameAttribute = channelNameAttribute;
}
public void setChannelMapping(ChannelMapping channelMapping) {
Assert.notNull(channelMapping, "channelMapping must not be null");
this.channelMapping = channelMapping;
}
private void createAdvisor() {
if (this.channelMapping == null) {
throw new IllegalStateException("channelMapping is required");
}
this.advisor = new PublisherAnnotationAdvisor(this.publisherAnnotationType, this.channelNameAttribute,
this.channelMapping);
}
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
return bean;
}
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
Class<?> targetClass = bean instanceof Advised ?
((Advised) bean).getTargetSource().getTargetClass() : bean.getClass();
if (targetClass == null) {
return bean;
}
if (advisor == null) {
createAdvisor();
}
if (AopUtils.canApply(this.advisor, targetClass)) {
if (bean instanceof Advised) {
((Advised) bean).addAdvisor(this.advisor);
return bean;
}
else {
ProxyFactory pf = new ProxyFactory(bean);
pf.addAdvisor(this.advisor);
return pf.getProxy(this.beanClassLoader);
}
}
else {
return bean;
}
}
}

View File

@@ -0,0 +1,104 @@
/*
* Copyright 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.config;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.framework.Advised;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.endpoint.GenericMessageEndpoint;
import org.springframework.integration.endpoint.MessageHandlerAdapter;
import org.springframework.integration.endpoint.annotation.Subscriber;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;
/**
* A {@link BeanPostProcessor} that creates a method-invoking handler adapter
* when it discovers methods annotated with {@link Subscriber @Subscriber}.
*
* @author Mark Fisher
*/
public class SubscriberAnnotationPostProcessor implements BeanPostProcessor {
private Log logger = LogFactory.getLog(this.getClass());
private Class<? extends Annotation> subscriberAnnotationType = Subscriber.class;
private String channelNameAttribute = "channel";
private MessageBus messageBus;
public void setSubscriberAnnotationType(Class<? extends Annotation> subscriberAnnotationType) {
Assert.notNull(subscriberAnnotationType, "subscriberAnnotationType must not be null");
this.subscriberAnnotationType = subscriberAnnotationType;
}
public void setChannelNameAttribute(String channelNameAttribute) {
Assert.notNull(channelNameAttribute, "channelNameAttribute must not be null");
this.channelNameAttribute = channelNameAttribute;
}
public void setMessageBus(MessageBus messageBus) {
Assert.notNull(messageBus, "messageBus must not be null");
this.messageBus = messageBus;
}
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
return bean;
}
public Object postProcessAfterInitialization(final Object bean, String beanName) throws BeansException {
final Class<?> targetClass = bean instanceof Advised ?
((Advised) bean).getTargetSource().getTargetClass() : bean.getClass();
if (targetClass == null) {
return bean;
}
ReflectionUtils.doWithMethods(targetClass, new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = method.getAnnotation(subscriberAnnotationType);
if (annotation != null) {
String channelName = (String) AnnotationUtils.getValue(annotation, channelNameAttribute);
MessageHandlerAdapter adapter = new MessageHandlerAdapter();
adapter.setMethod(method.getName());
adapter.setObject(bean);
adapter.afterPropertiesSet();
GenericMessageEndpoint endpoint = new GenericMessageEndpoint();
endpoint.setInputChannelName(channelName);
endpoint.setChannelMapping(messageBus);
endpoint.setHandler(adapter);
String endpointName = ClassUtils.getShortNameAsProperty(targetClass) +
"-" + method.getName() + "-endpoint";
messageBus.registerEndpoint(endpointName, endpoint);
if (logger.isInfoEnabled()) {
logger.info("registered endpoint: " + endpointName);
}
}
}
});
return bean;
}
}

View File

@@ -0,0 +1,121 @@
<?xml version="1.0" encoding="UTF-8"?>
<xsd:schema xmlns="http://www.springframework.org/schema/integration"
xmlns:xsd="http://www.w3.org/2001/XMLSchema"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:tool="http://www.springframework.org/schema/tool"
targetNamespace="http://www.springframework.org/schema/integration"
elementFormDefault="qualified"
attributeFormDefault="unqualified">
<xsd:import namespace="http://www.springframework.org/schema/beans"/>
<xsd:import namespace="http://www.springframework.org/schema/tool"/>
<xsd:annotation>
<xsd:documentation><![CDATA[
Defines the configuration elements for Spring Integration.
]]></xsd:documentation>
</xsd:annotation>
<xsd:element name="message-bus">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines a message bus.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="auto-create-channels" type="xsd:boolean"/>
<xsd:attribute name="invalid-message-channel" type="xsd:string"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="annotation-driven">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Enables the publisher annotation post-processor.
</xsd:documentation>
</xsd:annotation>
</xsd:complexType>
</xsd:element>
<xsd:element name="channel">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines a message channel.
</xsd:documentation>
</xsd:annotation>
<xsd:complexContent>
<xsd:extension base="beans:identifiedType">
<xsd:attribute name="capacity" type="xsd:integer"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="inbound-channel-adapter">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines an inbound channel adapter.
</xsd:documentation>
</xsd:annotation>
<xsd:complexContent>
<xsd:extension base="beans:identifiedType">
<xsd:attribute name="ref" type="xsd:string" use="required"/>
<xsd:attribute name="method" type="xsd:string" use="required"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="outbound-channel-adapter">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines an outbound channel adapter.
</xsd:documentation>
</xsd:annotation>
<xsd:complexContent>
<xsd:extension base="beans:identifiedType">
<xsd:attribute name="ref" type="xsd:string" use="required"/>
<xsd:attribute name="method" type="xsd:string" use="required"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="endpoint">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines a message endpoint.
</xsd:documentation>
</xsd:annotation>
<xsd:complexContent>
<xsd:extension base="beans:identifiedType">
<xsd:sequence>
<xsd:element ref="consumer" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="input-channel" type="xsd:string"/>
<xsd:attribute name="default-output-channel" type="xsd:string"/>
<xsd:attribute name="handler-ref" type="xsd:string"/>
<xsd:attribute name="handler-method" type="xsd:string"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="consumer">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines a consumer policy.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="period" type="xsd:int"/>
</xsd:complexType>
</xsd:element>
</xsd:schema>

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", 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;
}

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

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

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

View File

@@ -0,0 +1,130 @@
/*
* 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.MessageHandlingException;
import org.springframework.integration.MessagingConfigurationException;
import org.springframework.integration.bus.ConsumerPolicy;
import org.springframework.integration.channel.ChannelMapping;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
/**
* A generic endpoint implementation designed to accommodate a variety of
* strategies including:
* <ul>
* <li><i>source channel-adapter:</i> source adapter + target channel</li>
* <li><i>target channel-adapter:</i> source channel + target adapter</li>
* <li><i>one-way:</i> source + handler that returns null and no target</li>
* <li><i>request-reply:</i> source + handler and either a reply channel
* specified on the request message or a default target on the endpoint</i>
* </ul>
*
* @author Mark Fisher
*/
public class GenericMessageEndpoint implements MessageEndpoint {
private String inputChannelName;
private String defaultOutputChannelName;
private MessageHandler handler;
private ChannelMapping channelMapping;
private ConsumerPolicy consumerPolicy = new ConsumerPolicy();
/**
* Set the name of the channel from which this endpoint receives messages.
*/
public void setInputChannelName(String inputChannelName) {
this.inputChannelName = inputChannelName;
}
/**
* Return the name of the channel from which this endpoint receives messages.
*/
public String getInputChannelName() {
return this.inputChannelName;
}
public void setConsumerPolicy(ConsumerPolicy consumerPolicy) {
this.consumerPolicy = consumerPolicy;
}
public ConsumerPolicy getConsumerPolicy() {
return this.consumerPolicy;
}
/**
* Set the name of the channel to which this endpoint can send reply messages by default.
*/
public void setDefaultOutputChannelName(String defaultOutputChannelName) {
this.defaultOutputChannelName = defaultOutputChannelName;
}
/**
* Set a handler to be invoked for each consumed message.
*/
public void setHandler(MessageHandler handler) {
this.handler = handler;
}
/**
* Set the channel mapping to use for looking up channels by name.
*/
public void setChannelMapping(ChannelMapping channelMapping) {
this.channelMapping = channelMapping;
}
public void messageReceived(Message<?> message) {
if (this.handler == null) {
if (this.defaultOutputChannelName == null) {
throw new MessagingConfigurationException(
"endpoint must have either a 'handler' or 'defaultOutputChannelName'");
}
MessageChannel replyChannel = this.channelMapping.getChannel(this.defaultOutputChannelName);
replyChannel.send(message);
return;
}
Message<?> replyMessage = handler.handle(message);
if (replyMessage != null) {
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 'defaultOutputChannelName' "
+ "on the message endpoint.");
}
replyChannel.send(replyMessage);
}
}
private MessageChannel resolveReplyChannel(Message<?> message) {
if (this.channelMapping == null) {
return null;
}
String replyChannelName = message.getHeader().getReplyChannelName();
if (replyChannelName != null && replyChannelName.trim().length() > 0) {
return this.channelMapping.getChannel(replyChannelName);
}
return this.channelMapping.getChannel(this.defaultOutputChannelName);
}
}

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.");
}
}
}
}

View File

@@ -0,0 +1,42 @@
/*
* 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.bus.ConsumerPolicy;
import org.springframework.integration.channel.ChannelMapping;
import org.springframework.integration.message.Message;
/**
* Base interface for message endpoints.
*
* @author Mark Fisher
*/
public interface MessageEndpoint {
void setInputChannelName(String inputChannelName);
String getInputChannelName();
void setDefaultOutputChannelName(String defaultOutputChannelName);
ConsumerPolicy getConsumerPolicy();
void setChannelMapping(ChannelMapping channelMapping);
void messageReceived(Message<?> message);
}

View File

@@ -0,0 +1,78 @@
/*
* Copyright 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.endpoint;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.Ordered;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageMapper;
import org.springframework.integration.message.SimplePayloadMessageMapper;
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, Ordered, InitializingBean {
private T object;
private String method;
private MessageMapper mapper = new SimplePayloadMessageMapper();
private SimpleMethodInvoker<T> invoker;
private int order = Integer.MAX_VALUE;
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 setOrder(int order) {
this.order = order;
}
public int getOrder() {
return this.order;
}
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;
}
}

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

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

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

View File

@@ -0,0 +1,34 @@
/*
* 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.annotation;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import org.springframework.integration.handler.MessageHandler;
/**
* A strategy for programmatically creating a {@link MessageHandler} based on
* metadata provided by an annotation.
*
* @author Mark Fisher
*/
public interface AnnotationHandlerCreator {
MessageHandler createHandler(Object object, Method method, Annotation annotation);
}

View File

@@ -0,0 +1,49 @@
/*
* 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.annotation;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.annotation.Order;
import org.springframework.integration.endpoint.MessageHandlerAdapter;
import org.springframework.integration.handler.MessageHandler;
/**
* Default implementation of the handler creator strategy that creates a
* {@link MessageHandlerAdapter} for the provided object and method. This
* version does not even consider the annotation itself. It does however
* respect an {@link Order} annotation if present.
*
* @author Mark Fisher
*/
public class DefaultAnnotationHandlerCreator implements AnnotationHandlerCreator {
public MessageHandler createHandler(Object object, Method method, Annotation annotation) {
MessageHandlerAdapter<Object> adapter = new MessageHandlerAdapter<Object>();
adapter.setObject(object);
adapter.setMethod(method.getName());
Order orderAnnotation = (Order) AnnotationUtils.getAnnotation(method, Order.class);
if (orderAnnotation != null) {
adapter.setOrder(orderAnnotation.value());
}
adapter.afterPropertiesSet();
return adapter;
}
}

View File

@@ -0,0 +1,41 @@
/*
* 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.annotation;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import org.springframework.integration.message.Message;
/**
* Indicates that a method is capable of sending messages. The method must
* accept a single parameter that is either a {@link Message} or an Object to
* be passed as a message payload. The enclosing class should be annotated with
* {@link MessageEndpoint @MessageEndpoint}.
*
* @author Mark Fisher
*/
@java.lang.annotation.Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface DefaultOutput {
}

View File

@@ -0,0 +1,39 @@
/*
* 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.annotation;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
* Indicates that a method is capable of handling a message or message payload.
* The method may only accept a single parameter, and the enclosing class should
* be annotated with {@link MessageEndpoint @MessageEndpoint}.
*
* @author Mark Fisher
*/
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface Handler {
}

View File

@@ -0,0 +1,46 @@
/*
* 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.annotation;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import org.springframework.stereotype.Component;
/**
* Indicates that a class is capable of serving as a message endpoint.
*
* @author Mark Fisher
*/
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
@Component
public @interface MessageEndpoint {
String input() default "";
String defaultOutput() default "";
int pollPeriod() default 0;
}

View File

@@ -0,0 +1,179 @@
/*
* 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.annotation;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.core.OrderComparator;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.bus.ConsumerPolicy;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.endpoint.GenericMessageEndpoint;
import org.springframework.integration.endpoint.InboundMethodInvokingChannelAdapter;
import org.springframework.integration.endpoint.OutboundMethodInvokingChannelAdapter;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.MessageHandlerChain;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
/**
* A {@link BeanPostProcessor} implementation that generates endpoints for
* classes annotated with {@link MessageEndpoint @MessageEndpoint}.
*
* @author Mark Fisher
*/
public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor, InitializingBean {
private Map<Class<? extends Annotation>, AnnotationHandlerCreator> handlerCreators =
new ConcurrentHashMap<Class<? extends Annotation>, AnnotationHandlerCreator>();
private MessageBus messageBus;
public void setMessageBus(MessageBus messageBus) {
Assert.notNull(messageBus, "messageBus must not be null");
this.messageBus = messageBus;
}
public void setCustomHandlerCreators(
Map<Class<? extends Annotation>, AnnotationHandlerCreator> customHandlerCreators) {
for (Map.Entry<Class<? extends Annotation>, AnnotationHandlerCreator> entry : customHandlerCreators.entrySet()) {
this.handlerCreators.put(entry.getKey(), entry.getValue());
}
}
public void afterPropertiesSet() {
Assert.notNull(this.messageBus, "messageBus is required");
this.handlerCreators.put(Handler.class, new DefaultAnnotationHandlerCreator());
}
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
return bean;
}
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
MessageEndpoint endpointAnnotation = bean.getClass().getAnnotation(MessageEndpoint.class);
if (endpointAnnotation == null) {
return bean;
}
GenericMessageEndpoint endpoint = new GenericMessageEndpoint();
this.configureInputChannel(bean, beanName, endpointAnnotation, endpoint);
this.configureDefaultOutputChannel(bean, beanName, endpointAnnotation, endpoint);
MessageHandlerChain handlerChain = this.createHandlerChain(bean);
if (handlerChain != null) {
endpoint.setHandler(handlerChain);
}
this.messageBus.registerEndpoint(beanName, endpoint);
return endpoint;
}
private void configureInputChannel(final Object bean, final String beanName,
MessageEndpoint annotation, final GenericMessageEndpoint endpoint) {
String channelName = annotation.input();
if (StringUtils.hasText(channelName)) {
endpoint.setInputChannelName(channelName);
ConsumerPolicy consumerPolicy = new ConsumerPolicy();
consumerPolicy.setPeriod(annotation.pollPeriod());
endpoint.setConsumerPolicy(consumerPolicy);
return;
}
ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, Polled.class);
if (annotation != null) {
InboundMethodInvokingChannelAdapter<Object> adapter = new InboundMethodInvokingChannelAdapter<Object>();
adapter.setObject(bean);
adapter.setMethod(method.getName());
adapter.afterPropertiesSet();
String channelName = beanName + "-inputChannel";
messageBus.registerChannel(channelName, adapter);
endpoint.setInputChannelName(channelName);
int period = ((Polled) annotation).period();
endpoint.getConsumerPolicy().setPeriod(period);
if (period > 0) {
endpoint.getConsumerPolicy().setConcurrency(1);
endpoint.getConsumerPolicy().setMaxConcurrency(1);
endpoint.getConsumerPolicy().setMaxMessagesPerTask(1);
}
return;
}
}
});
}
private void configureDefaultOutputChannel(final Object bean, final String beanName,
final MessageEndpoint annotation, final GenericMessageEndpoint endpoint) {
String channelName = annotation.defaultOutput();
if (StringUtils.hasText(channelName)) {
endpoint.setDefaultOutputChannelName(channelName);
return;
}
ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, DefaultOutput.class);
if (annotation != null) {
OutboundMethodInvokingChannelAdapter<Object> adapter = new OutboundMethodInvokingChannelAdapter<Object>();
adapter.setObject(bean);
adapter.setMethod(method.getName());
adapter.afterPropertiesSet();
String channelName = beanName + "-defaultOutputChannel";
messageBus.registerChannel(channelName, adapter);
endpoint.setDefaultOutputChannelName(channelName);
return;
}
}
});
}
@SuppressWarnings("unchecked")
private MessageHandlerChain createHandlerChain(final Object bean) {
final List<MessageHandler> handlers = new ArrayList<MessageHandler>();
ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
for (Class<? extends Annotation> annotationType : handlerCreators.keySet()) {
Annotation annotation = AnnotationUtils.getAnnotation(method, annotationType);
if (annotation != null) {
MessageHandler handler = handlerCreators.get(annotationType).createHandler(bean, method, annotation);
if (handler != null) {
handlers.add(handler);
}
}
}
}
});
if (handlers.size() > 0) {
MessageHandlerChain handlerChain = new MessageHandlerChain();
Collections.sort(handlers, new OrderComparator());
for (MessageHandler handler : handlers) {
handlerChain.add(handler);
}
return handlerChain;
}
return null;
}
}

View File

@@ -0,0 +1,42 @@
/*
* 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.annotation;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
* Indicates that a method is capable of providing messages. The method must not
* accept any parameters but can return either a single object or collection.
* The enclosing class should be annotated with
* {@link MessageEndpoint @MessageEndpoint}.
*
* @author Mark Fisher
*/
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface Polled {
int period() default 1000;
}

View File

@@ -0,0 +1,40 @@
/*
* 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.annotation;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
* Indicates that a method-invoking handler adapter should delegate to this
* method.
*
* @author Mark Fisher
*/
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface Subscriber {
String channel();
}

View File

@@ -0,0 +1,110 @@
/*
* 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.file;
import java.io.File;
import java.io.FileFilter;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.FilenameFilter;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.endpoint.AbstractInboundChannelAdapter;
import org.springframework.util.Assert;
import org.springframework.util.FileCopyUtils;
/**
* A basic inbound channel adapter for polling a directory.
*
* @author Mark Fisher
*/
public class InboundFileAdapter extends AbstractInboundChannelAdapter {
private File directory;
private File copyToDirectory;
private FileFilter fileFilter;
private FilenameFilter filenameFilter;
private boolean isTextFile = true;
public InboundFileAdapter(File directory) {
Assert.notNull("directory must not be null");
this.directory = directory;
}
public void setCopyToDirectory(File copyToDirectory) {
this.copyToDirectory = copyToDirectory;
}
public void setFileFilter(FileFilter fileFilter) {
this.fileFilter = fileFilter;
}
public void setFilenameFilter(FilenameFilter filenameFilter) {
this.filenameFilter = filenameFilter;
}
public void setIsTextFile(boolean isTextFile) {
this.isTextFile = isTextFile;
}
@Override
protected Object doReceiveObject() {
System.out.println("polling");
File[] files = null;
if (this.fileFilter != null) {
files = this.directory.listFiles(fileFilter);
}
else if (this.filenameFilter != null) {
files = this.directory.listFiles(filenameFilter);
}
else {
files = this.directory.listFiles();
}
if (files == null) {
throw new MessageHandlingException("Problem occurred while polling for files. " +
"Is '" + directory.getAbsolutePath() + "' a directory?");
}
if (files.length == 0) {
return null;
}
File file = files[0];
try {
if (this.isTextFile) {
FileReader reader = new FileReader(file);
String result = FileCopyUtils.copyToString(reader);
if (this.copyToDirectory != null) {
FileWriter writer = new FileWriter(this.copyToDirectory.getAbsolutePath() + File.separator + file.getName());
FileCopyUtils.copy(new FileReader(file), writer);
}
file.delete();
return result;
}
return FileCopyUtils.copyToByteArray(file);
}
catch (Exception e) {
e.printStackTrace();
throw new MessageHandlingException("Failed to extract message", e);
}
}
}

View File

@@ -0,0 +1,70 @@
/*
* 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.jms;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import org.springframework.integration.MessagingConfigurationException;
import org.springframework.integration.endpoint.AbstractInboundChannelAdapter;
import org.springframework.jms.core.JmsTemplate;
/**
* A simple channel adapter for receiving JMS Messages with a polling listener.
*
* @author Mark Fisher
*/
public class JmsInboundAdapter extends AbstractInboundChannelAdapter {
private ConnectionFactory connectionFactory;
private Destination destination;
private JmsTemplate jmsTemplate;
public void setConnectionFactory(ConnectionFactory connectionFactory) {
this.connectionFactory = connectionFactory;
}
public void setDestination(Destination destination) {
this.destination = destination;
}
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
@Override
public void initialize() {
if (this.jmsTemplate == null) {
if (this.connectionFactory == null || this.destination == null) {
throw new MessagingConfigurationException("Either a 'jmsTemplate' or " +
"*both* 'connectionFactory' and 'destination' are required.");
}
this.jmsTemplate = new JmsTemplate();
this.jmsTemplate.setConnectionFactory(this.connectionFactory);
this.jmsTemplate.setDefaultDestination(this.destination);
}
}
@Override
protected Object doReceiveObject() {
return this.jmsTemplate.receiveAndConvert();
}
}

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.jms;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import org.springframework.integration.MessagingConfigurationException;
import org.springframework.integration.endpoint.AbstractOutboundChannelAdapter;
import org.springframework.jms.core.JmsTemplate;
/**
* A simple channel adapter for sending JMS Messages.
*
* @author Mark Fisher
*/
public class JmsOutboundAdapter extends AbstractOutboundChannelAdapter {
private ConnectionFactory connectionFactory;
private Destination destination;
private JmsTemplate jmsTemplate;
public void setConnectionFactory(ConnectionFactory connectionFactory) {
this.connectionFactory = connectionFactory;
}
public void setDestination(Destination destination) {
this.destination = destination;
}
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
@Override
public void initialize() {
if (this.jmsTemplate == null) {
if (this.connectionFactory == null || this.destination == null) {
throw new MessagingConfigurationException("Either a 'jmsTemplate' or " +
"*both* 'connectionFactory' and 'destination' are required.");
}
this.jmsTemplate = new JmsTemplate();
this.jmsTemplate.setConnectionFactory(this.connectionFactory);
this.jmsTemplate.setDefaultDestination(this.destination);
}
}
@Override
protected boolean doSendObject(Object object) throws Exception {
this.jmsTemplate.convertAndSend(object);
return true;
}
}

View File

@@ -0,0 +1,51 @@
/*
* 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.handler;
import org.springframework.integration.message.Message;
import org.springframework.util.Assert;
/**
* A message handler implementation that intercepts calls to another handler.
*
* @author Mark Fisher
*/
public abstract class InterceptingMessageHandler implements MessageHandler {
private MessageHandler target;
public InterceptingMessageHandler(MessageHandler target) {
Assert.notNull(target, "target must not be null");
this.target = target;
}
public Message handle(Message message) {
return handle(message, this.target);
}
/**
* The handler method for subclasses to implement.
*
* @param message the message to handle
* @param target the intercepted handler
* @return a reply message or null
*/
public abstract Message handle(Message message, MessageHandler target);
}

View File

@@ -0,0 +1,32 @@
/*
* 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.handler;
import org.springframework.integration.message.Message;
/**
* Generic message handler interface. Typical implementations will translate
* between the generic Messages of the integration framework and the domain
* objects that are passed-to and returned-from business components.
*
* @author Mark Fisher
*/
public interface MessageHandler {
Message<?> handle(Message<?> message);
}

View File

@@ -0,0 +1,59 @@
/*
* 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.handler;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import org.springframework.integration.message.Message;
/**
* A message handler implementation that passes incoming messages through a
* chain of handlers.
*
* @author Mark Fisher
*/
public class MessageHandlerChain implements MessageHandler {
private List<MessageHandler> handlers = new CopyOnWriteArrayList<MessageHandler>();
/**
* Add a handler to the end of the chain.
*/
public void add(MessageHandler handler) {
this.handlers.add(handler);
}
/**
* Add a handler to the chain at the specified index.
*/
public void add(int index, MessageHandler handler) {
this.handlers.add(index, handler);
}
public Message handle(Message message) {
for (MessageHandler next : handlers) {
message = next.handle(message);
if (message == null) {
return null;
}
}
return message;
}
}

View File

@@ -0,0 +1,96 @@
/*
* Copyright 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.message;
import java.util.concurrent.locks.ReentrantLock;
import org.springframework.integration.SystemInterruptedException;
import org.springframework.integration.transformer.ObjectTransformer;
import org.springframework.util.Assert;
/**
* Base Message class defining common properties such as id, header, and lock.
*
* @author Mark Fisher
*/
public class GenericMessage<T> implements Message<T> {
private Object id;
private MessageHeader header = new MessageHeader();
private T payload;
private ReentrantLock lock;
public GenericMessage(Object id, T payload) {
Assert.notNull(id, "id must not be null");
Assert.notNull(payload, "payload must not be null");
this.id = id;
this.payload = payload;
}
public Object getId() {
return this.id;
}
public MessageHeader getHeader() {
return this.header;
}
protected void setHeader(MessageHeader header) {
this.header = header;
}
public T getPayload() {
return this.payload;
}
protected void setPayload(T newPayload) {
this.payload = newPayload;
}
public void lock() {
this.lock.lock();
}
public void lockInterruptibly() {
try {
lock.lockInterruptibly();
}
catch (InterruptedException e) {
throw new SystemInterruptedException("Unable to obtain lock for message", e);
}
}
public void unlock() {
lock.unlock();
}
public void transformPayload(ObjectTransformer transformer) {
this.lock();
try {
// TODO: remove this method (probably) or parameterize transformer
this.setPayload((T) transformer.transform(this.getPayload()));
}
finally {
this.unlock();
}
}
}

View File

@@ -0,0 +1,40 @@
/*
* Copyright 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.message;
import org.springframework.integration.transformer.ObjectTransformer;
/**
* The central interface that any Message type must implement.
*
* @author Mark Fisher
*/
public interface Message<T> {
Object getId();
MessageHeader getHeader();
T getPayload();
void transformPayload(ObjectTransformer transformer);
void lock();
void unlock();
}

View File

@@ -0,0 +1,114 @@
/*
* Copyright 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.message;
import java.util.Date;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
/**
* A holder for Message metadata. This includes information that is used by the
* messaging system (such <i>correlationId</i>) as well as information that is
* relevant for specific messaging endpoints. For the latter, String values may
* be stored as <i>properties</i> and Object values may be stored as
* <i>attributes</i>.
*
* @author Mark Fisher
*/
public class MessageHeader {
private Object correlationId;
private String replyChannelName;
private Date expiration;
private int sequenceNumber = 1;
private int sequenceSize = 1;
private Properties properties = new Properties();
private Map<String, Object> attributes = new ConcurrentHashMap<String, Object>();
public Object getCorrelationId() {
return this.correlationId;
}
public void setCorrelationId(Object correlationId) {
this.correlationId = correlationId;
}
public String getReplyChannelName() {
return this.replyChannelName;
}
public void setReplyChannelName(String replyChannelName) {
this.replyChannelName = replyChannelName;
}
public int getSequenceNumber() {
return this.sequenceNumber;
}
/**
* Set the expiration date for this message or <code>null</code> to
* indicate 'never expire'. The default is <code>null</code>.
*/
public void setExpiration(Date expiration) {
this.expiration = expiration;
}
/**
* Return the expiration date for this message or <code>null</code> to
* indicate 'never expire'.
*/
public Date getExpiration() {
return this.expiration;
}
public void setSequenceNumber(int sequenceNumber) {
this.sequenceNumber = sequenceNumber;
}
public int getSequenceSize() {
return this.sequenceSize;
}
public void setSequenceSize(int sequenceSize) {
this.sequenceSize = sequenceSize;
}
public String getProperty(String key) {
return this.properties.getProperty(key);
}
public void setProperty(String key, String value) {
this.properties.setProperty(key, value);
}
public Object getAttribute(String key) {
return this.attributes.get(key);
}
public void setAttribute(String key, Object value) {
this.attributes.put(key, value);
}
}

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.message;
/**
* Strategy interface for mapping between messages and objects.
*
* @author Mark Fisher
*/
public interface MessageMapper<M,O> {
/**
* Map to a {@link Message} from the given object.
*/
Message<M> toMessage(O source);
/**
* Map from the given {@link Message} to an object.
*/
O fromMessage(Message<M> message);
}

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.message;
/**
* Strategy interface for message selection.
*
* @author Mark Fisher
*/
public interface MessageSelector {
boolean accept(Message message);
}

View File

@@ -0,0 +1,68 @@
/*
* Copyright 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.message;
import org.springframework.integration.util.RandomGuidUidGenerator;
import org.springframework.integration.util.UidGenerator;
import org.springframework.util.Assert;
/**
* A {@link MessageMapper} implementation that simply wraps and unwraps a
* payload object in a {@link DocumentMessage}.
*
* @author Mark Fisher
*/
public class SimplePayloadMessageMapper<T> implements MessageMapper<T,T> {
private UidGenerator uidGenerator;
/**
* Create a mapper with the provided id generation strategy.
*
* @param uidGenerator the generator to use for message ids
*/
public SimplePayloadMessageMapper(UidGenerator uidGenerator) {
Assert.notNull(uidGenerator, "uidGenerator must not be null");
this.uidGenerator = uidGenerator;
}
/**
* Create a mapper with the default id generation strategy.
*
* @see RandomGuidUidGenerator
*/
public SimplePayloadMessageMapper() {
this.uidGenerator = new RandomGuidUidGenerator();
}
/**
* Return the payload of the given Message.
*/
public T fromMessage(Message<T> message) {
return message.getPayload();
}
/**
* Return a {@link DocumentMessage} with the given object as its payload.
*/
public Message<T> toMessage(T source) {
return new GenericMessage<T>(uidGenerator.generateUid(), source);
}
}

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.message;
/**
* @author Mark Fisher
*/
public class StringMessage extends GenericMessage<String> {
public StringMessage(Object id, String payload) {
super(id, payload);
}
}

View File

@@ -0,0 +1,30 @@
/*
* 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.router;
import java.util.Collection;
/**
* Base interface for aggregators.
*
* @author Mark Fisher
*/
public interface Aggregator<T> {
T aggregate(Collection<T> t);
}

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.router;
/**
* Strategy interface for content-based routing to a channel name.
*
* @author Mark Fisher
*/
public interface ChannelNameResolver<T> {
String resolve(T t);
}

View File

@@ -0,0 +1,30 @@
/*
* 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.router;
import org.springframework.integration.channel.MessageChannel;
/**
* Strategy interface for content-based routing to a channel instance.
*
* @author Mark Fisher
*/
public interface ChannelResolver<T> {
MessageChannel resolve(T t);
}

View File

@@ -0,0 +1,31 @@
/*
* 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.router;
import java.util.Collection;
import java.util.List;
/**
* Base interface for resequencers.
*
* @author Mark Fisher
*/
public interface Resequencer<T> {
List<T> resequence(Collection<T> t);
}

View File

@@ -0,0 +1,30 @@
/*
* 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.router;
/**
* Base strategy interface for barriers. Defines the common requirement of
* routing scenarios that involve waiting for a particular condition to be
* satisfied, such as an {@link Aggregator} or {@link Resequencer}.
*
* @author Mark Fisher
*/
public interface RoutingBarrier<T> {
boolean checkCondition(T t);
}

View File

@@ -0,0 +1,30 @@
/*
* 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.router;
import java.util.Collection;
/**
* Base interface for splitters.
*
* @author Mark Fisher
*/
public interface Splitter<T> {
Collection<T> split(T t);
}

View File

@@ -0,0 +1,31 @@
/*
* 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.samples;
import org.springframework.integration.endpoint.annotation.Subscriber;
/**
* @author Mark Fisher
*/
public class Logger {
@Subscriber(channel="quotes")
public void log(Object o) {
System.out.println(o);
}
}

View File

@@ -0,0 +1,39 @@
/*
* 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.samples;
import java.math.BigDecimal;
/**
* @author Mark Fisher
*/
public class Quote {
private String ticker;
private BigDecimal price;
public Quote(String ticker, BigDecimal price) {
this.ticker = ticker;
this.price = price;
}
public String toString() {
return ticker + ": " + price;
}
}

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.samples;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.Random;
import org.springframework.integration.endpoint.annotation.MessageEndpoint;
import org.springframework.integration.endpoint.annotation.Polled;
/**
* @author Mark Fisher
*/
@MessageEndpoint(defaultOutput="quotes")
public class QuotePublisher {
@Polled(period=300)
public Quote getQuote() {
BigDecimal price = new BigDecimal(new Random().nextDouble() * 100);
return new Quote("SOA", price.setScale(2, RoundingMode.HALF_EVEN));
}
}

View File

@@ -0,0 +1,32 @@
/*
* 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.samples;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
/**
* @author Mark Fisher
*/
public class StockQuoteDemo {
public static void main(String[] args) throws Exception {
AbstractApplicationContext context = new ClassPathXmlApplicationContext("stockQuoteDemo.xml", StockQuoteDemo.class);
context.start();
System.in.read();
}
}

View File

@@ -0,0 +1,20 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<message-bus/>
<annotation-driven/>
<channel id="quotes"/>
<beans:bean id="quotePublisher" class="org.springframework.integration.samples.QuotePublisher"/>
<beans:bean id="logger" class="org.springframework.integration.samples.Logger"/>
</beans:beans>

View File

@@ -0,0 +1,45 @@
/*
* 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.transformer;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageSelector;
/**
* Handler for deciding whether to pass a message. Implements
* {@link MessageHandler} and simply delegates to a {@link MessageSelector}.
*
* @author Mark Fisher
*/
public class MessageFilter implements MessageHandler {
private MessageSelector selector;
public MessageFilter(MessageSelector selector) {
this.selector = selector;
}
public Message handle(Message message) {
if (this.selector.accept(message)) {
return message;
}
return null;
}
}

View File

@@ -0,0 +1,26 @@
/*
* 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.transformer;
/**
* @author Mark Fisher
*/
public interface ObjectTransformer {
public Object transform(Object source);
}

View File

@@ -0,0 +1,197 @@
/*
* 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.util;
/*
* RandomGUID from http://www.javaexchange.com/aboutRandomGUID.html
* @version 1.2.1 11/05/02 @author Marc A. Mnich
*
* From www.JavaExchange.com, Open Software licensing
*
* 11/05/02 -- Performance enhancement from Mike Dubman. Moved InetAddr.getLocal to static block. Mike has measured a 10
* fold improvement in run time. 01/29/02 -- Bug fix: Improper seeding of nonsecure Random object caused duplicate GUIDs
* to be produced. Random object is now only created once per JVM. 01/19/02 -- Modified random seeding and added new
* constructor to allow secure random feature. 01/14/02 -- Added random function seeding with JVM run time
*/
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.security.SecureRandom;
import java.util.Random;
/**
* Globally unique identifier generator.
* <p>
* In the multitude of java GUID generators, I found none that guaranteed randomness. GUIDs are guaranteed to be
* globally unique by using ethernet MACs, IP addresses, time elements, and sequential numbers. GUIDs are not expected
* to be random and most often are easy/possible to guess given a sample from a given generator. SQL Server, for example
* generates GUID that are unique but sequencial within a given instance.
* <p>
* GUIDs can be used as security devices to hide things such as files within a filesystem where listings are unavailable
* (e.g. files that are served up from a Web server with indexing turned off). This may be desirable in cases where
* standard authentication is not appropriate. In this scenario, the RandomGuids are used as directories. Another
* example is the use of GUIDs for primary keys in a database where you want to ensure that the keys are secret. Random
* GUIDs can then be used in a URL to prevent hackers (or users) from accessing records by guessing or simply by
* incrementing sequential numbers.
* <p>
* There are many other possibilities of using GUIDs in the realm of security and encryption where the element of
* randomness is important. This class was written for these purposes but can also be used as a general purpose GUID
* generator as well.
* <p>
* RandomGuid generates truly random GUIDs by using the system's IP address (name/IP), system time in milliseconds (as
* an integer), and a very large random number joined together in a single String that is passed through an MD5 hash.
* The IP address and system time make the MD5 seed globally unique and the random number guarantees that the generated
* GUIDs will have no discernible pattern and cannot be guessed given any number of previously generated GUIDs. It is
* generally not possible to access the seed information (IP, time, random number) from the resulting GUIDs as the MD5
* hash algorithm provides one way encryption.
* <p>
* <b>Security of RandomGuid</b>: RandomGuid can be called one of two ways -- with the basic java Random number
* generator or a cryptographically strong random generator (SecureRandom). The choice is offered because the secure
* random generator takes about 3.5 times longer to generate its random numbers and this performance hit may not be
* worth the added security especially considering the basic generator is seeded with a cryptographically strong random
* seed.
* <p>
* Seeding the basic generator in this way effectively decouples the random numbers from the time component making it
* virtually impossible to predict the random number component even if one had absolute knowledge of the System time.
* Thanks to Ashutosh Narhari for the suggestion of using the static method to prime the basic random generator.
* <p>
* Using the secure random option, this class complies with the statistical random number generator tests specified in
* FIPS 140-2, Security Requirements for Cryptographic Modules, section 4.9.1.
* <p>
* I converted all the pieces of the seed to a String before handing it over to the MD5 hash so that you could print it
* out to make sure it contains the data you expect to see and to give a nice warm fuzzy. If you need better
* performance, you may want to stick to byte[] arrays.
* <p>
* I believe that it is important that the algorithm for generating random GUIDs be open for inspection and
* modification. This class is free for all uses.
*
* @version 1.2.1 11/05/02
* @author Marc A. Mnich
*/
public class RandomGuid {
private static Random random;
private static SecureRandom secureRandom;
private static String id;
private String guid;
/*
* Static block to take care of one time secureRandom seed. It takes a few seconds to initialize SecureRandom. You
* might want to consider removing this static block or replacing it with a "time since first loaded" seed to reduce
* this time. This block will run only once per JVM instance.
*/
static {
secureRandom = new SecureRandom();
long secureInitializer = secureRandom.nextLong();
random = new Random(secureInitializer);
try {
id = InetAddress.getLocalHost().toString();
} catch (UnknownHostException e) {
throw new RuntimeException(e);
}
}
/**
* Default constructor. With no specification of security option, this constructor defaults to lower security, high
* performance.
*/
public RandomGuid() {
getRandomGuid(false);
}
/**
* Constructor with security option. Setting secure true enables each random number generated to be
* cryptographically strong. Secure false defaults to the standard Random function seeded with a single
* cryptographically strong random number.
*/
public RandomGuid(boolean secure) {
getRandomGuid(secure);
}
/**
* Method to generate the random GUID.
*/
private void getRandomGuid(boolean secure) {
MessageDigest md5 = null;
StringBuffer sbValueBeforeMD5 = new StringBuffer();
try {
md5 = MessageDigest.getInstance("MD5");
} catch (NoSuchAlgorithmException e) {
throw new RuntimeException(e);
}
long time = System.currentTimeMillis();
long rand = 0;
if (secure) {
rand = secureRandom.nextLong();
} else {
rand = random.nextLong();
}
// This StringBuffer can be as long as you need; the MD5
// hash will always return 128 bits. You can change
// the seed to include anything you want here.
// You could even stream a file through the MD5 making
// the odds of guessing it at least as great as that
// of guessing the contents of the file!
sbValueBeforeMD5.append(id);
sbValueBeforeMD5.append(":");
sbValueBeforeMD5.append(Long.toString(time));
sbValueBeforeMD5.append(":");
sbValueBeforeMD5.append(Long.toString(rand));
String valueBeforeMD5 = sbValueBeforeMD5.toString();
md5.update(valueBeforeMD5.getBytes());
byte[] array = md5.digest();
StringBuffer sb = new StringBuffer();
for (int j = 0; j < array.length; ++j) {
int b = array[j] & 0xFF;
if (b < 0x10)
sb.append('0');
sb.append(Integer.toHexString(b));
}
guid = sb.toString();
}
/**
* Convert to the standard format for GUID (Useful for SQL Server UniqueIdentifiers, etc). Example:
* "C2FEEEAC-CFCD-11D1-8B05-00600806D9B6".
*/
public String toString() {
String raw = guid.toUpperCase();
StringBuffer sb = new StringBuffer();
sb.append(raw.substring(0, 8));
sb.append("-");
sb.append(raw.substring(8, 12));
sb.append("-");
sb.append(raw.substring(12, 16));
sb.append("-");
sb.append(raw.substring(16, 20));
sb.append("-");
sb.append(raw.substring(20));
return sb.toString();
}
}

View File

@@ -0,0 +1,59 @@
/*
* 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.util;
import java.io.Serializable;
/**
* An id generator that uses the RandomGuid support class. The default
* implementation used by the integration system.
*
* @author Keith Donald
*/
@SuppressWarnings("serial")
public class RandomGuidUidGenerator implements UidGenerator, Serializable {
/**
* Should the random GUID generated be secure?
*/
private boolean secure;
/**
* Returns whether or not the generated random numbers are <i>secure</i>,
* meaning cryptographically strong.
*/
public boolean isSecure() {
return secure;
}
/**
* Sets whether or not the generated random numbers should be <i>secure</i>.
* If set to true, generated GUIDs are cryptographically strong.
*/
public void setSecure(boolean secure) {
this.secure = secure;
}
public Serializable generateUid() {
return new RandomGuid(secure).toString();
}
public Serializable parseUid(String encodedUid) {
return encodedUid;
}
}

View File

@@ -0,0 +1,42 @@
/*
* 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.util;
import java.io.Serializable;
/**
* A strategy for generating ids to uniquely identify integration artifacts such
* as Messages.
*
* @author Keith Donald
*/
public interface UidGenerator {
/**
* Generate a new unique id.
* @return a serializable id, guaranteed to be unique in some context
*/
public Serializable generateUid();
/**
* Convert the string-encoded uid into its original object form.
* @param encodedUid the string encoded uid
* @return the converted uid
*/
public Serializable parseUid(String encodedUid);
}