From 164023c8bc617a51517d3797d5f5b9e75a5e8e5e Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 29 Jul 2010 21:14:24 +0000 Subject: [PATCH] INT-1301 MessagingTemplate methods now return void instead of boolean, but they throw Exceptions if the underlying channel send() invocation returns false. --- .../integration/core/MessagingOperations.java | 6 ++--- .../integration/core/MessagingTemplate.java | 27 +++++++++---------- .../endpoint/MessageProducerSupport.java | 4 +-- .../endpoint/SourcePollingChannelAdapter.java | 5 ++-- .../integration/filter/MessageFilter.java | 15 +++-------- .../gateway/AbstractMessagingGateway.java | 4 +-- .../AbstractReplyProducingMessageHandler.java | 11 +++----- .../integration/handler/DelayHandler.java | 12 ++++++--- .../router/AbstractMessageRouter.java | 16 +++++++---- .../MessageTransformingHandler.java | 11 +++----- ...elatingMessageHandlerIntegrationTests.java | 4 +++ ...nEvaluatingMessageGroupProcessorTests.java | 18 +++++++++++++ ...ApplicationEventInboundChannelAdapter.java | 4 +-- 13 files changed, 76 insertions(+), 61 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingOperations.java b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingOperations.java index eeec505491..dabcdfef88 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingOperations.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingOperations.java @@ -49,7 +49,7 @@ public interface MessagingOperations { * @param message the message to send * @throws MessagingException if an error occurs during message sending */ -

boolean send(Message

message) throws MessagingException; +

void send(Message

message) throws MessagingException; /** * Send a message to the specified channel. @@ -57,7 +57,7 @@ public interface MessagingOperations { * @param message the message to send * @throws MessagingException if an error occurs during message sending */ -

boolean send(MessageChannel channel, Message

message) throws MessagingException; +

void send(MessageChannel channel, Message

message) throws MessagingException; /** * Send a message to the specified channel. @@ -66,7 +66,7 @@ public interface MessagingOperations { * @param message the message to send * @throws MessagingException if an error occurs during message sending */ - //TODO:

boolean send(String channelName, Message

message) throws MessagingException; + //TODO:

void send(String channelName, Message

message) throws MessagingException; //------------------------------------------------------------------------- diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java index 6fdabc1373..4321cc7b16 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java @@ -162,20 +162,21 @@ public class MessagingTemplate implements MessagingOperations, InitializingBean } } - public

boolean send(final Message

message) { - return this.send(this.getRequiredDefaultChannel(), message); + public

void send(final Message

message) { + this.send(this.getRequiredDefaultChannel(), message); } - public

boolean send(final MessageChannel channel, final Message

message) { + public

void send(final MessageChannel channel, final Message

message) { TransactionTemplate txTemplate = this.getTransactionTemplate(); if (txTemplate != null) { - return txTemplate.execute(new TransactionCallback() { - public Boolean doInTransaction(TransactionStatus status) { - return doSend(channel, message); + txTemplate.execute(new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + doSend(channel, message); + return null; } }); } - return this.doSend(channel, message); + this.doSend(channel, message); } public

Message

receive() { @@ -213,16 +214,16 @@ public class MessagingTemplate implements MessagingOperations, InitializingBean return this.doSendAndReceive(channel, request); } - private boolean doSend(MessageChannel channel, Message message) { + private void doSend(MessageChannel channel, Message message) { Assert.notNull(channel, "channel must not be null"); long timeout = this.sendTimeout; boolean sent = (timeout >= 0) ? channel.send(message, timeout) : channel.send(message); - if (!sent && this.logger.isTraceEnabled()) { - this.logger.trace("failed to send message to channel '" + channel + "' within timeout: " + timeout); + if (!sent) { + throw new MessageDeliveryException(message, + "failed to send message to channel '" + channel + "' within timeout: " + timeout); } - return sent; } @SuppressWarnings("unchecked") @@ -246,9 +247,7 @@ public class MessagingTemplate implements MessagingOperations, InitializingBean .setReplyChannel(replyChannel) .setErrorChannel(replyChannel) .build(); - if (!this.doSend(channel, request)) { - throw new MessageDeliveryException(request, "failed to send message to channel"); - } + this.doSend(channel, request); Message reply = this.doReceive(replyChannel); if (reply != null) { reply = MessageBuilder.fromMessage(reply) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java index e49882aa33..23e75321ee 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java @@ -48,11 +48,11 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements Assert.notNull(this.outputChannel, "outputChannel is required"); } - protected boolean sendMessage(Message message) { + protected void sendMessage(Message message) { if (message != null) { message.getHeaders().getHistory().addEvent(this); } - return this.messagingTemplate.send(this.outputChannel, message); + this.messagingTemplate.send(this.outputChannel, message); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java index d5bb009e6a..521d914d53 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2010 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. @@ -75,7 +75,8 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint { protected boolean doPoll() { Message message = this.source.receive(); if (message != null) { - return this.messagingTemplate.send(this.outputChannel, message); + this.messagingTemplate.send(this.outputChannel, message); + return true; } return false; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java b/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java index 3b087822b0..79d43ba0de 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java @@ -18,7 +18,6 @@ package org.springframework.integration.filter; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.integration.Message; -import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.MessageHeaders; import org.springframework.integration.MessageRejectedException; import org.springframework.integration.core.MessageChannel; @@ -105,11 +104,7 @@ public class MessageFilter extends AbstractReplyProducingMessageHandler { return message; } if (this.discardChannel != null) { - boolean discarded = this.sendReplyMessage(message, this.discardChannel); - if (!discarded) { - throw new MessageDeliveryException(message, - "failed to send rejected Message to the discard channel"); - } + this.sendReplyMessage(message, this.discardChannel); } if (this.throwExceptionOnRejection) { throw new MessageRejectedException(message); @@ -117,11 +112,9 @@ public class MessageFilter extends AbstractReplyProducingMessageHandler { return null; } + @Override protected void handleResult(Object replyMessage, MessageHeaders requestHeaders, MessageChannel replyChannel) { - if (!this.sendReplyMessage((Message) replyMessage, replyChannel)) { - throw new MessageDeliveryException((Message) replyMessage, - "failed to send reply Message to channel '" + replyChannel + "'. Consider increasing the " + - "send timeout of this endpoint."); - } + this.sendReplyMessage((Message) replyMessage, replyChannel); } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java index 35745ab809..fb4e273f07 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java @@ -144,9 +144,7 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { "send is not supported, because no request channel has been configured"); Message message = this.toMessage(object); Assert.notNull(message, "message must not be null"); - if (!this.messagingTemplate.send(this.requestChannel, message)) { - throw new MessageDeliveryException(message, "failed to send Message to channel"); - } + this.messagingTemplate.send(this.requestChannel, message); } protected Object receive() { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java index 85a3d82e96..fc6f147217 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java @@ -17,7 +17,6 @@ package org.springframework.integration.handler; import org.springframework.integration.Message; -import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.MessageHandlingException; import org.springframework.integration.MessageHeaders; import org.springframework.integration.core.ChannelResolver; @@ -106,11 +105,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa protected void handleResult(Object result, MessageHeaders requestHeaders, MessageChannel replyChannel) { Message replyMessage = this.createReplyMessage(result, requestHeaders); - if (!this.sendReplyMessage(replyMessage, replyChannel)) { - throw new MessageDeliveryException(replyMessage, - "failed to send reply Message to channel '" + replyChannel + "'. Consider increasing the " + - "send timeout of this endpoint."); - } + this.sendReplyMessage(replyMessage, replyChannel); } @SuppressWarnings("unchecked") @@ -124,11 +119,11 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa return builder.build(); } - protected boolean sendReplyMessage(Message replyMessage, MessageChannel replyChannel) { + protected void sendReplyMessage(Message replyMessage, MessageChannel replyChannel) { if (logger.isDebugEnabled()) { logger.debug("handler '" + this + "' sending reply Message: " + replyMessage); } - return this.messagingTemplate.send(replyChannel, replyMessage); + this.messagingTemplate.send(replyChannel, replyMessage); } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java index 9de3f710cb..8482f963cd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java @@ -241,13 +241,17 @@ public class DelayHandler extends IntegrationObjectSupport implements MessageHan MessageChannel errorChannel = resolveErrorChannelIfPossible(message); if (errorChannel != null) { ErrorMessage errorMessage = new ErrorMessage(exception); - boolean sent = messagingTemplate.send(errorChannel, errorMessage); - if (!sent && logger.isWarnEnabled()) { - logger.warn("Failed to send MessageDeliveryException to error channel.", exception); + try { + messagingTemplate.send(errorChannel, errorMessage); + } + catch (Exception e2) { + if (logger.isWarnEnabled()) { + logger.warn("Failed to send MessagingException to error channel.", exception); + } } } else if (logger.isWarnEnabled()) { - logger.warn("No error channel available. MessageDeliveryException will be ignored.", exception); + logger.warn("No error channel available. MessagingException will be ignored.", exception); } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java index 094001f420..6dc3aca40d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java @@ -22,6 +22,7 @@ import java.util.UUID; import org.springframework.integration.Message; import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.MessageHeaders; +import org.springframework.integration.MessagingException; import org.springframework.integration.core.MessageBuilder; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessagingTemplate; @@ -116,19 +117,24 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler { .setHeader(MessageHeaders.ID, UUID.randomUUID()) .build(); if (channel != null) { - if (this.messagingTemplate.send(channel, messageToSend)) { + try { + this.messagingTemplate.send(channel, messageToSend); sent = true; } - else if (!this.ignoreSendFailures) { - throw new MessageDeliveryException(message, - "Router failed to send to channel: " + channel); + catch (MessagingException e) { + if (!this.ignoreSendFailures) { + throw e; + } + else if (this.logger.isDebugEnabled()) { + this.logger.debug(e); + } } } } } if (!sent) { if (this.defaultOutputChannel != null) { - sent = this.messagingTemplate.send(this.defaultOutputChannel, message); + this.messagingTemplate.send(this.defaultOutputChannel, message); } else if (this.resolutionRequired) { throw new MessageDeliveryException(message, diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java index c8b2ded25f..4e2bab8b2d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java @@ -18,7 +18,6 @@ package org.springframework.integration.transformer; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.integration.Message; -import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.MessageHeaders; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageHandler; @@ -72,12 +71,10 @@ public class MessageTransformingHandler extends AbstractReplyProducingMessageHan throw new MessageTransformationException(message, e); } } - + + @Override protected void handleResult(Object replyMessage, MessageHeaders requestHeaders, MessageChannel replyChannel) { - if (!this.sendReplyMessage((Message) replyMessage, replyChannel)) { - throw new MessageDeliveryException((Message) replyMessage, - "failed to send reply Message to channel '" + replyChannel + "'. Consider increasing the " + - "send timeout of this endpoint."); - } + this.sendReplyMessage((Message) replyMessage, replyChannel); } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.java index 1d980a50ba..4ed0c8cb1d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.java @@ -16,12 +16,15 @@ package org.springframework.integration.aggregator; +import static org.mockito.Mockito.isA; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import org.junit.Before; import org.junit.Test; + import org.springframework.integration.Message; import org.springframework.integration.core.MessageBuilder; import org.springframework.integration.core.MessageChannel; @@ -40,6 +43,7 @@ public class CorrelatingMessageHandlerIntegrationTests { @Before public void setupHandler() { + when(outputChannel.send(isA(Message.class))).thenReturn(true); defaultHandler.setOutputChannel(outputChannel); defaultHandler.setSendTimeout(-1); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessorTests.java index 9512e2ac33..e0de459664 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessorTests.java @@ -1,6 +1,23 @@ +/* + * Copyright 2002-2010 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.aggregator; import static org.junit.matchers.JUnitMatchers.hasItems; +import static org.mockito.Matchers.isA; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -48,6 +65,7 @@ public class ExpressionEvaluatingMessageGroupProcessorTests { @Before public void setup() { + when(outputChannel.send(isA(Message.class))).thenReturn(true); messages.clear(); for (int i = 0; i < 5; i++) { messages.add(MessageBuilder.withPayload(i + 1).setHeader("foo", "bar").build()); diff --git a/spring-integration-event/src/main/java/org/springframework/integration/event/ApplicationEventInboundChannelAdapter.java b/spring-integration-event/src/main/java/org/springframework/integration/event/ApplicationEventInboundChannelAdapter.java index c442f9c72d..efc0fef211 100644 --- a/spring-integration-event/src/main/java/org/springframework/integration/event/ApplicationEventInboundChannelAdapter.java +++ b/spring-integration-event/src/main/java/org/springframework/integration/event/ApplicationEventInboundChannelAdapter.java @@ -63,8 +63,8 @@ public class ApplicationEventInboundChannelAdapter extends MessageProducerSuppor } } - private boolean sendEventAsMessage(ApplicationEvent event) { - return this.sendMessage(MessageBuilder.withPayload(event).build()); + private void sendEventAsMessage(ApplicationEvent event) { + this.sendMessage(MessageBuilder.withPayload(event).build()); } @Override