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 new file mode 100644 index 0000000000..eeec505491 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingOperations.java @@ -0,0 +1,212 @@ +/* + * 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.core; + +import org.springframework.integration.Message; +import org.springframework.integration.MessagingException; + +/** + * Specifies a basic set of messaging operations. + * + *

Implemented by {@link MessagingTemplate}. Even though most calling code + * will depend on the template directly (e.g. to access setter methods), this + * interface is a useful option to enhance testability, as it can easily be mocked + * or stubbed. + * + *

Defines a variety of methods for sending and receiving {@link Message}s + * across {@link MessageChannel}s including the use of converters where necessary. + * Convenience methods also support sending and receiving based on channel name, + * where the template will delegate to its {@link ChannelResolver} to locate the + * actual {@link MessageChannel} instance. + * + * @author Mark Fisher + * @since 2.0 + * @see MessagingTemplate + */ +public interface MessagingOperations { + + //------------------------------------------------------------------------- + // Convenience methods for sending messages + //------------------------------------------------------------------------- + + /** + * Send a message to the default channel. + *

This will only work with a default channel specified! + * @param message the message to send + * @throws MessagingException if an error occurs during message sending + */ +

boolean send(Message

message) throws MessagingException; + + /** + * Send a message to the specified channel. + * @param channel the channel to which the message will be sent + * @param message the message to send + * @throws MessagingException if an error occurs during message sending + */ +

boolean send(MessageChannel channel, Message

message) throws MessagingException; + + /** + * Send a message to the specified channel. + * @param channelName the name of the channel to which the message will be sent + * (to be resolved to an actual channel by a ChannelResolver) + * @param message the message to send + * @throws MessagingException if an error occurs during message sending + */ + //TODO:

boolean send(String channelName, Message

message) throws MessagingException; + + + //------------------------------------------------------------------------- + // Convenience methods for sending auto-converted messages + //------------------------------------------------------------------------- + + // TODO: convert and send methods... + + /** + * Send the given object to the default channel, converting the object + * to a message with a configured MessageConverter. + *

This will only work with a default channel specified! + * @param message the object to convert to a message + * @throws MessagingException if an error occurs + */ + //void convertAndSend(Object message) throws MessagingException; + + /** + * Send the given object to the specified channel, converting the object + * to a message with a configured MessageConverter. + * @param channel the channel to send this message to + * @param message the object to convert to a message + * @throws MessagingException if an error occurs + */ + //void convertAndSend(MessageChannel channel, Object message) throws MessagingException; + + /** + * Send the given object to the specified channel, converting the object + * to a message with a configured MessageConverter. + * @param channelName the name of the channel to send this message to + * (to be resolved to an actual channel by a ChannelResolver) + * @param message the object to convert to a message + * @throws MessagingException if an error occurs + */ + //void convertAndSend(String destinationName, Object message) throws MessagingException; + + /** + * Send the given object to the default destination, converting the object + * to a JMS message with a configured MessageConverter. The MessagePostProcessor + * callback allows for modification of the message after conversion. + *

This will only work with a default destination specified! + * @param message the object to convert to a message + * @param postProcessor the callback to modify the message + * @throws JmsException checked JMSException converted to unchecked + */ + //void convertAndSend(Object message, MessagePostProcessor postProcessor) throws MessagingException; + + /** + * TODO: either define a MessagePostProcessor that accepts a builder or accept HeaderMapper here + */ + //void convertAndSend(MessageChannel channel, Object message, MessagePostProcessor postProcessor) throws MessagingException; + + /** + * TODO: see above + */ + //void convertAndSend(String channelName, Object message, MessagePostProcessor postProcessor) throws MessagingException; + + + //------------------------------------------------------------------------- + // Convenience methods for receiving messages + //------------------------------------------------------------------------- + + /** + * Receive a message synchronously from the default channel, but only + * wait up to a specified time for delivery. + *

This method should be used carefully, since it will block the thread + * until the message becomes available or until the timeout value is exceeded. + *

This will only work with a default channel specified! + * @return the message received from the default channel or null if the timeout expires + * @throws MessagingException if an error occurs during message reception + */ +

Message

receive() throws MessagingException; + + /** + * Receive a message synchronously from the specified channel, but only + * wait up to a specified time for delivery. + *

This method should be used carefully, since it will block the thread + * until the message becomes available or until the timeout value is exceeded. + * @param channel the channel from which a message should be received + * @return the message received from the channel or null if the timeout expires + * @throws MessagingException if an error occurs during message reception + */ +

Message

receive(PollableChannel channel) throws MessagingException; + + /** + * Receive a message synchronously from the specified channel, but only + * wait up to a specified time for delivery. + *

This method should be used carefully, since it will block the thread + * until the message becomes available or until the timeout value is exceeded. + * @param channelName the name of the channel from which a message should be received + * (to be resolved to an actual channel by a ChannelResolver) + * @return the message received from the channel or null if the timeout expires + * @throws MessagingException if an error occurs during message reception + */ + // TODO:

Message

receive(String channelName) throws MessagingException; + + + // TODO: receiveSelected(selector), receiveSelected(channel, selector), receiveSelected(channelName, selector) ? + + //------------------------------------------------------------------------- + // Convenience methods for receiving auto-converted messages + //------------------------------------------------------------------------- + + // TODO: receive and convert methods... + + /** + * Receive a message synchronously from the default channel, but only + * wait up to a specified time for delivery. Convert the message into an + * object with a configured MessageConverter. + *

This method should be used carefully, since it will block the thread + * until the message becomes available or until the timeout value is exceeded. + *

This will only work with a default channel specified! + * @return the message received from the channel or null if the timeout expires. + * @throws MessagingException if an error occurs during message reception + */ + //Object receiveAndConvert() throws JmsException; + + /** + * Receive a message synchronously from the specified channel, but only + * wait up to a specified time for delivery. Convert the message into an + * object with a configured MessageConverter. + *

This method should be used carefully, since it will block the thread + * until the message becomes available or until the timeout value is exceeded. + * @param channel the channel from which a message should be received + * @return the message received from the channel or null if the timeout expires. + * @throws MessagingException if an error occurs during message reception + */ + //Object receiveAndConvert(PollableChannel channel) throws JmsException; + + /** + * Receive a message synchronously from the specified channel, but only + * wait up to a specified time for delivery. Convert the message into an + * object with a configured MessageConverter. + *

This method should be used carefully, since it will block the thread + * until the message becomes available or until the timeout value is exceeded. + * @param channelName the name of the channel from which a message should be received + * (to be resolved to an actual channel by a ChannelResolver) + * @return the message received from the channel or null if the timeout expires. + * @throws MessagingException if an error occurs during message reception + */ + //Object receiveAndConvert(String channelName) throws JmsException; + +} 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 4ec74c0da3..6fdabc1373 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 @@ -45,7 +45,7 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class MessagingTemplate implements InitializingBean { +public class MessagingTemplate implements MessagingOperations, InitializingBean { protected final Log logger = LogFactory.getLog(this.getClass()); @@ -162,11 +162,11 @@ public class MessagingTemplate implements InitializingBean { } } - public boolean send(final Message message) { + public

boolean send(final Message

message) { return this.send(this.getRequiredDefaultChannel(), message); } - public boolean send(final MessageChannel channel, final Message message) { + public

boolean send(final MessageChannel channel, final Message

message) { TransactionTemplate txTemplate = this.getTransactionTemplate(); if (txTemplate != null) { return txTemplate.execute(new TransactionCallback() { @@ -178,18 +178,18 @@ public class MessagingTemplate implements InitializingBean { return this.doSend(channel, message); } - public Message receive() { + public

Message

receive() { MessageChannel channel = this.getRequiredDefaultChannel(); Assert.state(channel instanceof PollableChannel, "The 'defaultChannel' must be a PollableChannel for receive operations."); return this.receive((PollableChannel) channel); } - public Message receive(final PollableChannel channel) { + public

Message

receive(final PollableChannel channel) { TransactionTemplate txTemplate = this.getTransactionTemplate(); if (txTemplate != null) { - return txTemplate.execute(new TransactionCallback>() { - public Message doInTransaction(TransactionStatus status) { + return txTemplate.execute(new TransactionCallback>() { + public Message

doInTransaction(TransactionStatus status) { return doReceive(channel); } }); @@ -225,7 +225,8 @@ public class MessagingTemplate implements InitializingBean { return sent; } - private Message doReceive(PollableChannel channel) { + @SuppressWarnings("unchecked") + private

Message

doReceive(PollableChannel channel) { Assert.notNull(channel, "channel must not be null"); long timeout = this.receiveTimeout; Message message = (timeout >= 0) @@ -234,7 +235,7 @@ public class MessagingTemplate implements InitializingBean { if (message == null && this.logger.isTraceEnabled()) { this.logger.trace("failed to receive message from channel '" + channel + "' within timeout: " + timeout); } - return message; + return (Message

) message; } private Message doSendAndReceive(MessageChannel channel, Message request) { diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java index a7bb2e7cf9..6730088af5 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java @@ -124,8 +124,7 @@ public class JdbcPollingChannelAdapterParserTests { return null; } }); - @SuppressWarnings("unchecked") - Message> message = (Message>) messagingTemplate.receive(); + Message> message = messagingTemplate.receive(); assertNotNull(message); assertEquals(2, message.getPayload().size()); }