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

Added the MessageProducer interface and refactored AbstractMessageBarrierConsumer to no longer extend AbstractReplyProducingMessageConsumer.

Browse Source
This commit is contained in:
Mark Fisher
2008-10-13 20:59:40 +00:00
parent 4de01bc09b
commit 3565609d1e
4 changed files with 112 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

93
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierConsumer.java
View File

@@ -30,11 +30,12 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.endpoint.AbstractReplyProducingMessageConsumer;
import org.springframework.integration.endpoint.ReplyMessageHolder;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.endpoint.AbstractMessageConsumer;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageProducer;
import org.springframework.integration.scheduling.IntervalTrigger;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.integration.scheduling.TaskSchedulerAware;
@@ -64,8 +65,10 @@ import org.springframework.util.ObjectUtils;
* @author Mark Fisher
* @author Marius Bogoevici
*/
public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProducingMessageConsumer
implements TaskSchedulerAware, InitializingBean {
public abstract class AbstractMessageBarrierConsumer extends AbstractMessageConsumer
implements MessageProducer, TaskSchedulerAware, InitializingBean {
public final static long DEFAULT_SEND_TIMEOUT = 1000;
public final static long DEFAULT_TIMEOUT = 60000;
@@ -73,8 +76,13 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc
public final static int DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY = 1000;
protected final Log logger = LogFactory.getLog(this.getClass());
private MessageChannel outputChannel;
private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate();
private volatile MessageChannel discardChannel;
protected final ConcurrentMap<Object, MessageBarrier> barriers =
@@ -97,6 +105,15 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc
private ScheduledFuture<?> reaperFutureTask;
public AbstractMessageBarrierConsumer() {
this.channelTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT);
}
public void setOutputChannel(MessageChannel outputChannel) {
this.outputChannel = outputChannel;
}
/**
* Specify a channel for sending Messages that arrive after their aggregation
* group has either completed or timed-out.
@@ -138,6 +155,10 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc
this.timeout = timeout;
}
public void setSendTimeout(long sendTimeout) {
this.channelTemplate.setSendTimeout(sendTimeout);
}
public void setTaskScheduler(TaskScheduler taskScheduler) {
this.taskScheduler = taskScheduler;
}
@@ -167,7 +188,7 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc
}
@Override
protected final void onMessage(Message<?> message, ReplyMessageHolder replyHolder) {
protected final void onMessageInternal(Message<?> message) {
if (!this.initialized) {
this.afterPropertiesSet();
}
@@ -177,36 +198,47 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc
this.getClass().getSimpleName() + " requires the 'correlationId' property");
}
if (this.trackedCorrelationIds.contains(correlationId)) {
if (logger.isDebugEnabled()) {
logger.debug("Handling of Message group with correlationId '"
+ correlationId + "' has already completed or timed out.");
}
this.sendToDiscardChannelIfAvailable(message);
return;
this.discardMessage(message, correlationId);
}
else {
this.processMessage(message, correlationId);
}
}
private void discardMessage(Message<?> message, Object correlationId) {
if (logger.isDebugEnabled()) {
logger.debug("Handling of Message group with correlationId '"
+ correlationId + "' has already completed or timed out.");
}
if (this.discardChannel != null) {
boolean sent = this.channelTemplate.send(message, this.discardChannel);
if (!sent && logger.isWarnEnabled()) {
logger.warn("unable to send to 'discardChannel', message: " + message);
}
}
}
private void processMessage(Message<?> message, Object correlationId) {
MessageBarrier barrier = barriers.putIfAbsent(correlationId, createMessageBarrier());
if (barrier == null) {
barrier = barriers.get(correlationId);
}
List<Message<?>> releasedMessages = barrier.addAndRelease(message);
if (CollectionUtils.isEmpty(releasedMessages)) {
return;
if (!CollectionUtils.isEmpty(releasedMessages)) {
if (isBarrierRemovable(correlationId, releasedMessages)) {
this.removeBarrier(correlationId);
}
Message<?>[] processedMessages = this.processReleasedMessages(correlationId, releasedMessages);
if (!ObjectUtils.isEmpty(processedMessages)) {
this.afterRelease(correlationId, releasedMessages);
}
}
if (isBarrierRemovable(correlationId, releasedMessages)) {
this.removeBarrier(correlationId);
}
Message<?>[] processedMessages = this.processReleasedMessages(correlationId, releasedMessages);
if (ObjectUtils.isEmpty(processedMessages)) {
return;
}
this.afterRelease(correlationId, releasedMessages);
return;
}
private void afterRelease(Object correlationId, List<Message<?>> releasedMessages) {
Message<?>[] processedMessages = this.processReleasedMessages(correlationId, releasedMessages);
for (Message<?> result : processedMessages) {
MessageChannel replyChannel = this.getOutputChannel();
MessageChannel replyChannel = this.outputChannel;
if (replyChannel == null) {
replyChannel = this.resolveReplyChannelFromMessage(result);
if (replyChannel == null) {
@@ -214,7 +246,7 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc
}
}
if (replyChannel != null) {
this.sendReplyMessage(result, replyChannel);
this.channelTemplate.send(result, replyChannel);
}
else if (logger.isWarnEnabled()) {
logger.warn("unable to determine reply target for aggregation result: " + result);
@@ -222,17 +254,6 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc
}
}
private void sendToDiscardChannelIfAvailable(Message<?> message) {
if (this.discardChannel != null) {
boolean sent = this.sendReplyMessage(message, this.discardChannel);
if (!sent) {
if (logger.isWarnEnabled()) {
logger.warn("unable to send to 'discardChannel', message: " + message);
}
}
}
}
protected MessageChannel resolveReplyChannelFromMessage(Message<?> message) {
Object returnAddress = message.getHeaders().getReturnAddress();
if (returnAddress != null) {
@@ -273,7 +294,7 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc
}
else {
for (Message<?> message : messages) {
sendToDiscardChannelIfAvailable(message);
discardMessage(message, correlationId);
}
}
}

45
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
View File

@@ -19,6 +19,7 @@ package org.springframework.integration.config.annotation;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.generic.GenericBeanFactoryAccessor;
@@ -29,12 +30,11 @@ import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.channel.SubscribableChannel;
import org.springframework.integration.endpoint.AbstractMessageConsumer;
import org.springframework.integration.endpoint.AbstractReplyProducingMessageConsumer;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.PollingConsumerEndpoint;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageProducer;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -91,34 +91,33 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
if (StringUtils.hasText(inputChannelName)) {
MessageChannel inputChannel = this.channelResolver.resolveChannelName(inputChannelName);
Assert.notNull(inputChannel, "failed to resolve inputChannel '" + inputChannelName + "'");
if (consumer instanceof AbstractMessageConsumer) {
if (inputChannel instanceof PollableChannel) {
PollingConsumerEndpoint pollingEndpoint = new PollingConsumerEndpoint(
consumer, (PollableChannel) inputChannel);
if (pollerAnnotation != null) {
AnnotationConfigUtils.configurePollingEndpointWithPollerAnnotation(
pollingEndpoint, pollerAnnotation, this.beanFactoryAccessor.getBeanFactory());
}
endpoint = pollingEndpoint;
}
else if (inputChannel instanceof SubscribableChannel) {
Assert.isTrue(pollerAnnotation == null,
"The @Poller annotation should only be provided for a PollableChannel");
endpoint = new SubscribingConsumerEndpoint(consumer, (SubscribableChannel) inputChannel);
}
else {
throw new IllegalArgumentException("unsupported channel type: ["
+ inputChannel.getClass() + "]");
if (inputChannel instanceof PollableChannel) {
PollingConsumerEndpoint pollingEndpoint = new PollingConsumerEndpoint(
consumer, (PollableChannel) inputChannel);
if (pollerAnnotation != null) {
AnnotationConfigUtils.configurePollingEndpointWithPollerAnnotation(
pollingEndpoint, pollerAnnotation, this.beanFactoryAccessor.getBeanFactory());
}
endpoint = pollingEndpoint;
}
if (consumer instanceof AbstractReplyProducingMessageConsumer) {
else if (inputChannel instanceof SubscribableChannel) {
Assert.isTrue(pollerAnnotation == null,
"The @Poller annotation should only be provided for a PollableChannel");
endpoint = new SubscribingConsumerEndpoint(consumer, (SubscribableChannel) inputChannel);
}
else {
throw new IllegalArgumentException("unsupported channel type: [" + inputChannel.getClass() + "]");
}
if (consumer instanceof MessageProducer) {
String outputChannelName = (String) AnnotationUtils.getValue(annotation, OUTPUT_CHANNEL_ATTRIBUTE);
if (StringUtils.hasText(outputChannelName)) {
MessageChannel outputChannel = this.channelResolver.resolveChannelName(outputChannelName);
Assert.notNull(outputChannel, "unable to resolve outputChannel '" + outputChannelName + "'");
((AbstractReplyProducingMessageConsumer) consumer).setOutputChannel(outputChannel);
((MessageProducer) consumer).setOutputChannel(outputChannel);
}
((AbstractReplyProducingMessageConsumer) consumer).setChannelResolver(this.channelResolver);
}
if (consumer instanceof BeanFactoryAware) {
((BeanFactoryAware) consumer).setBeanFactory(this.beanFactoryAccessor.getBeanFactory());
}
}
return endpoint;

4
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractReplyProducingMessageConsumer.java
View File

@@ -26,6 +26,7 @@ import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageHeaders;
import org.springframework.integration.message.MessageProducer;
import org.springframework.integration.message.MessageRejectedException;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.selector.MessageSelector;
@@ -36,7 +37,8 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public abstract class AbstractReplyProducingMessageConsumer extends AbstractMessageConsumer implements BeanFactoryAware {
public abstract class AbstractReplyProducingMessageConsumer extends AbstractMessageConsumer
implements MessageProducer, BeanFactoryAware {
public static final long DEFAULT_SEND_TIMEOUT = 1000;

30
org.springframework.integration/src/main/java/org/springframework/integration/message/MessageProducer.java Normal file
View File

@@ -0,0 +1,30 @@
/*
* Copyright 2002-2008 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.message;
import org.springframework.integration.channel.MessageChannel;
/**
* Base interface for any component that produces Messages.
*
* @author Mark Fisher
*/
public interface MessageProducer {
void setOutputChannel(MessageChannel outputChannel);
}