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

INT-1301 MessagingTemplate methods now return void instead of boolean, but they throw Exceptions if the underlying channel send() invocation returns false.

Browse Source
This commit is contained in:
Mark Fisher
2010-07-29 21:14:24 +00:00
parent 00d1713981
commit 164023c8bc
13 changed files with 76 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-core/src/main/java/org/springframework/integration/core/MessagingOperations.java
View File

@@ -49,7 +49,7 @@ public interface MessagingOperations {
* @param message the message to send
* @throws MessagingException if an error occurs during message sending
*/
<P> boolean send(Message<P> message) throws MessagingException;
<P> void send(Message<P> 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
*/
<P> boolean send(MessageChannel channel, Message<P> message) throws MessagingException;
<P> void send(MessageChannel channel, Message<P> 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: <P> boolean send(String channelName, Message<P> message) throws MessagingException;
//TODO: <P> void send(String channelName, Message<P> message) throws MessagingException;
//-------------------------------------------------------------------------

27
spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java
View File

@@ -162,20 +162,21 @@ public class MessagingTemplate implements MessagingOperations, InitializingBean
}
}
public <P> boolean send(final Message<P> message) {
return this.send(this.getRequiredDefaultChannel(), message);
public <P> void send(final Message<P> message) {
this.send(this.getRequiredDefaultChannel(), message);
}
public <P> boolean send(final MessageChannel channel, final Message<P> message) {
public <P> void send(final MessageChannel channel, final Message<P> message) {
TransactionTemplate txTemplate = this.getTransactionTemplate();
if (txTemplate != null) {
return txTemplate.execute(new TransactionCallback<Boolean>() {
public Boolean doInTransaction(TransactionStatus status) {
return doSend(channel, message);
txTemplate.execute(new TransactionCallback<Object>() {
public Object doInTransaction(TransactionStatus status) {
doSend(channel, message);
return null;
}
});
}
return this.doSend(channel, message);
this.doSend(channel, message);
}
public <P> Message<P> 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)

4
spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java
View File

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

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

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

15
spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java
View File

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

4
spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java
View File

@@ -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() {

11
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java
View File

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

12
spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java
View File

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

16
spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java
View File

@@ -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,

11
spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java
View File

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

4
spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.java
View File

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

18
spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessorTests.java
View File

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

4
spring-integration-event/src/main/java/org/springframework/integration/event/ApplicationEventInboundChannelAdapter.java
View File

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