Add MessageSendingOperations for JMS

This commit adds a JMS implementation of MessageSendingOperations,
allowing to send JMS messages using Spring's standard Messaging
abstraction.

MessagingMessageConverter is a standard JMS's MessageConverter that
can convert Spring's Message to JMS message and vice versa. Existing
infrastructure has been updated to use this implementation.

Issue: SPR-11772

spring-jms/src/main/java/org/springframework/jms/listener/adapter/AbstractAdaptableMessageListener.java

@@ -32,6 +32,7 @@ import org.springframework.jms.support.JmsUtils;
 import org.springframework.jms.support.converter.JmsHeaderMapper;
 import org.springframework.jms.support.converter.MessageConversionException;
 import org.springframework.jms.support.converter.MessageConverter;
+import org.springframework.jms.support.converter.MessagingMessageConverter;
 import org.springframework.jms.support.converter.SimpleJmsHeaderMapper;
 import org.springframework.jms.support.converter.SimpleMessageConverter;
 import org.springframework.jms.support.destination.DestinationResolver;
@@ -60,7 +61,7 @@ public abstract class AbstractAdaptableMessageListener
 
 	private MessageConverter messageConverter;
 
-	private JmsHeaderMapper headerMapper = new SimpleJmsHeaderMapper();
+	private MessagingMessageConverterAdapter messagingMessageConverter = new MessagingMessageConverterAdapter();
 
 
 	/**
@@ -152,6 +153,10 @@ public abstract class AbstractAdaptableMessageListener
 		return this.messageConverter;
 	}
 
+	protected MessageConverter getMessagingMessageConverter() {
+		return this.messagingMessageConverter;
+	}
+
 	/**
 	 * Set the {@link JmsHeaderMapper} implementation to use to map the
 	 * standard JMS headers. By default {@link SimpleJmsHeaderMapper} is
@@ -160,15 +165,7 @@ public abstract class AbstractAdaptableMessageListener
 	 */
 	public void setHeaderMapper(JmsHeaderMapper headerMapper) {
 		Assert.notNull(headerMapper, "HeaderMapper must not be null");
-		this.headerMapper = headerMapper;
-	}
-
-	/**
-	 * Return the {@link JmsHeaderMapper} that converts headers from
-	 * and to the messaging abstraction.
-	 */
-	protected JmsHeaderMapper getHeaderMapper() {
-		return headerMapper;
+		this.messagingMessageConverter.setHeaderMapper(headerMapper);
 	}
 
 	/**
@@ -284,10 +281,7 @@ public abstract class AbstractAdaptableMessageListener
 		MessageConverter converter = getMessageConverter();
 		if (converter != null) {
 			if (result instanceof org.springframework.messaging.Message) {
-				org.springframework.messaging.Message<?> message = (org.springframework.messaging.Message<?>) result;
-				Message reply = converter.toMessage(message.getPayload(), session);
-				getHeaderMapper().fromHeaders(message.getHeaders(), reply);
-				return reply;
+				return messagingMessageConverter.toMessage(result, session);
 			}
 			else {
 				return converter.toMessage(result, session);
@@ -404,6 +398,18 @@ public abstract class AbstractAdaptableMessageListener
 	}
 
 
+	/**
+	 * Delegates payload extraction to {@link #extractMessage(javax.jms.Message)} to
+	 * enforce backward compatibility.
+	 */
+	private class MessagingMessageConverterAdapter extends MessagingMessageConverter {
+
+		@Override
+		protected Object extractPayload(Message message) throws JMSException {
+			return extractMessage(message);
+		}
+	}
+
 	/**
 	 * Internal class combining a destination name
 	 * and its target destination type (queue or topic).

spring-jms/src/main/java/org/springframework/jms/listener/adapter/MessagingMessageListenerAdapter.java

@@ -16,16 +16,14 @@
 
 package org.springframework.jms.listener.adapter;
 
-import java.util.Map;
-
 import javax.jms.JMSException;
 import javax.jms.Session;
 
 import org.springframework.jms.support.converter.JmsHeaderMapper;
+import org.springframework.jms.support.converter.MessageConversionException;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
-import org.springframework.messaging.support.MessageBuilder;
 
 /**
  * A {@link javax.jms.MessageListener} adapter that invokes a configurable
@@ -75,12 +73,12 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
 
 	@SuppressWarnings("unchecked")
 	protected Message<?> toMessagingMessage(javax.jms.Message jmsMessage) {
-		Map<String, Object> mappedHeaders = getHeaderMapper().toHeaders(jmsMessage);
-		Object convertedObject = extractMessage(jmsMessage);
-		MessageBuilder<Object> builder = (convertedObject instanceof org.springframework.messaging.Message) ?
-				MessageBuilder.fromMessage((org.springframework.messaging.Message<Object>) convertedObject) :
-				MessageBuilder.withPayload(convertedObject);
-		return builder.copyHeadersIfAbsent(mappedHeaders).build();
+		try {
+			return (Message<?>) getMessagingMessageConverter().fromMessage(jmsMessage);
+		}
+		catch (JMSException e) {
+			throw new MessageConversionException("Could not unmarshal message", e);
+		}
 	}
 
 	/**

spring-jms/src/main/java/org/springframework/jms/messaging/JmsMessageSendingOperations.java

@@ -0,0 +1,92 @@
+/*
+ * Copyright 2002-2014 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.jms.messaging;
+
+import java.util.Map;
+
+import javax.jms.Destination;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.core.MessagePostProcessor;
+import org.springframework.messaging.core.MessageSendingOperations;
+
+/**
+ * A specialization of {@link MessageSendingOperations} for JMS related
+ * operations that allows to specify a destination name rather than the
+ * actual {@link javax.jms.Destination}
+ *
+ * @author Stephane Nicoll
+ * @since 4.1
+ * @see org.springframework.jms.core.JmsTemplate
+ */
+public interface JmsMessageSendingOperations extends MessageSendingOperations<Destination> {
+
+	/**
+	 * Send a message to the given destination.
+	 * @param destinationName the name of the target destination
+	 * @param message the message to send
+	 */
+	void send(String destinationName, Message<?> message) throws MessagingException;
+
+	/**
+	 * Convert the given Object to serialized form, possibly using a
+	 * {@link org.springframework.messaging.converter.MessageConverter},
+	 * wrap it as a message and send it to the given destination.
+	 * @param destinationName the name of the target destination
+	 * @param payload the Object to use as payload
+	 */
+	void convertAndSend(String destinationName, Object payload) throws MessagingException;
+
+	/**
+	 * Convert the given Object to serialized form, possibly using a
+	 * {@link org.springframework.messaging.converter.MessageConverter},
+	 * wrap it as a message with the given headers and send it to
+	 * the given destination.
+	 * @param destinationName the name of the target destination
+	 * @param payload the Object to use as payload
+	 * @param headers headers for the message to send
+	 */
+	void convertAndSend(String destinationName, Object payload, Map<String, Object> headers)
+			throws MessagingException;
+
+	/**
+	 * Convert the given Object to serialized form, possibly using a
+	 * {@link org.springframework.messaging.converter.MessageConverter},
+	 * wrap it as a message, apply the given post processor, and send
+	 * the resulting message to the given destination.
+	 * @param destinationName the name of the target destination
+	 * @param payload the Object to use as payload
+	 * @param postProcessor the post processor to apply to the message
+	 */
+	void convertAndSend(String destinationName, Object payload, MessagePostProcessor postProcessor)
+			throws MessagingException;
+
+	/**
+	 * Convert the given Object to serialized form, possibly using a
+	 * {@link org.springframework.messaging.converter.MessageConverter},
+	 * wrap it as a message with the given headers, apply the given post processor,
+	 * and send the resulting message to the given destination.
+	 * @param destinationName the name of the target destination
+	 * @param payload the Object to use as payload
+	 * @param headers headers for the message to send
+	 * @param postProcessor the post processor to apply to the message
+	 */
+	void convertAndSend(String destinationName, Object payload, Map<String,
+			Object> headers, MessagePostProcessor postProcessor) throws MessagingException;
+
+}

spring-jms/src/main/java/org/springframework/jms/messaging/JmsMessagingTemplate.java

@@ -0,0 +1,206 @@
+/*
+ * Copyright 2002-2014 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.jms.messaging;
+
+import java.util.Map;
+
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Session;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.jms.core.JmsTemplate;
+import org.springframework.jms.core.MessageCreator;
+import org.springframework.jms.support.converter.MessageConverter;
+import org.springframework.jms.support.converter.MessagingMessageConverter;
+import org.springframework.jms.support.converter.SimpleJmsHeaderMapper;
+import org.springframework.jms.support.converter.SimpleMessageConverter;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.core.AbstractMessageSendingTemplate;
+import org.springframework.messaging.core.MessagePostProcessor;
+import org.springframework.util.Assert;
+
+/**
+ * An implementation of {@link JmsMessageSendingOperations}.
+ *
+ * @author Stephane Nicoll
+ * @since 4.1
+ */
+public class JmsMessagingTemplate
+		extends AbstractMessageSendingTemplate<Destination>
+		implements JmsMessageSendingOperations, InitializingBean {
+
+	private JmsTemplate jmsTemplate;
+
+	private MessageConverter jmsMessageConverter = new MessagingMessageConverter(
+			new SimpleMessageConverter(), new SimpleJmsHeaderMapper());
+
+	private String defaultDestinationName;
+
+	public JmsMessagingTemplate() {
+	}
+
+	/**
+	 * Create an instance with the {@link JmsTemplate} to use.
+	 */
+	public JmsMessagingTemplate(JmsTemplate jmsTemplate) {
+		Assert.notNull("JmsTemplate must not be null");
+		this.jmsTemplate = jmsTemplate;
+	}
+
+	/**
+	 * Set the {@link JmsTemplate} to use.
+	 */
+	public void setJmsTemplate(JmsTemplate jmsTemplate) {
+		this.jmsTemplate = jmsTemplate;
+	}
+
+	/**
+	 * Set the {@link MessageConverter} to use to convert a {@link Message} from
+	 * the messaging to and from a {@link javax.jms.Message}. By default, a
+	 * {@link MessagingMessageConverter} is defined using a {@link SimpleMessageConverter}
+	 * to convert the payload of the message.
+	 * <p>Consider configuring a {@link MessagingMessageConverter} with a different
+	 * {@link MessagingMessageConverter#setPayloadConverter(MessageConverter) payload converter}
+	 * for more advanced scenario.
+	 *
+	 * @see org.springframework.jms.support.converter.MessagingMessageConverter
+	 */
+	public void setJmsMessageConverter(MessageConverter jmsMessageConverter) {
+		this.jmsMessageConverter = jmsMessageConverter;
+	}
+
+	/**
+	 * Configure the default destination name to use in send methods that don't have
+	 * a destination argument. If a default destination is not configured, send methods
+	 * without a destination argument will raise an exception if invoked.
+	 * @see #setDefaultDestination(Object)
+	 */
+	public void setDefaultDestinationName(String defaultDestinationName) {
+		this.defaultDestinationName = defaultDestinationName;
+	}
+
+	/**
+	 * Return the configured default destination name.
+	 */
+	public String getDefaultDestinationName() {
+		return this.defaultDestinationName;
+	}
+
+	@Override
+	public void afterPropertiesSet() {
+		Assert.notNull(this.jmsTemplate, "Property 'jmsTemplate' is required");
+		Assert.notNull(this.jmsMessageConverter, "Property 'jmsMessageConverter' is required");
+	}
+
+
+	@Override
+	public void send(Message<?> message) {
+		Destination defaultDestination = getDefaultDestination();
+		if (defaultDestination != null) {
+			send(defaultDestination, message);
+		}
+		else {
+			send(getRequiredDefaultDestinationName(), message);
+		}
+	}
+
+	@Override
+	public void convertAndSend(Object payload) throws MessagingException {
+		convertAndSend(payload, null);
+	}
+
+	@Override
+	public void convertAndSend(Object payload, MessagePostProcessor postProcessor) throws MessagingException {
+		Destination defaultDestination = getDefaultDestination();
+		if (defaultDestination != null) {
+			convertAndSend(defaultDestination, payload, postProcessor);
+		}
+		else {
+			convertAndSend(getRequiredDefaultDestinationName(), payload, postProcessor);
+		}
+	}
+
+	@Override
+	public void send(String destinationName, Message<?> message) throws MessagingException {
+		doSend(destinationName, message);
+	}
+
+	@Override
+	public void convertAndSend(String destinationName, Object payload) throws MessagingException {
+		convertAndSend(destinationName, payload, (Map<String, Object>) null);
+	}
+
+	@Override
+	public void convertAndSend(String destinationName, Object payload, Map<String, Object> headers)
+			throws MessagingException {
+		convertAndSend(destinationName, payload, headers, null);
+	}
+
+	@Override
+	public void convertAndSend(String destinationName, Object payload, MessagePostProcessor postProcessor)
+			throws MessagingException {
+		convertAndSend(destinationName, payload, null, postProcessor);
+	}
+
+	@Override
+	public void convertAndSend(String destinationName, Object payload, Map<String, Object> headers,
+			MessagePostProcessor postProcessor) throws MessagingException {
+		Message<?> message = doConvert(payload, headers, postProcessor);
+		send(destinationName, message);
+	}
+
+	@Override
+	protected void doSend(Destination destination, Message<?> message) {
+		jmsTemplate.send(destination, new MessagingMessageCreator(message, this.jmsMessageConverter));
+	}
+
+	protected void doSend(String destinationName, Message<?> message) {
+		jmsTemplate.send(destinationName, new MessagingMessageCreator(message, this.jmsMessageConverter));
+	}
+
+	protected String getRequiredDefaultDestinationName() {
+		String name = getDefaultDestinationName();
+		if (name == null) {
+			throw new IllegalStateException(
+					"No 'defaultDestination' or 'defaultDestinationName' specified. " +
+							"Check configuration of JmsMessagingTemplate."
+			);
+		}
+		return name;
+	}
+
+
+	private static class MessagingMessageCreator implements MessageCreator {
+
+		private final Message<?> message;
+
+		private final MessageConverter messageConverter;
+
+		private MessagingMessageCreator(Message<?> message, MessageConverter messageConverter) {
+			this.message = message;
+			this.messageConverter = messageConverter;
+		}
+
+		@Override
+		public javax.jms.Message createMessage(Session session) throws JMSException {
+			return messageConverter.toMessage(message, session);
+		}
+	}
+
+}

spring-jms/src/main/java/org/springframework/jms/messaging/package-info.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2002-2014 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.
+ */
+
+/**
+ * JMS integration for Spring's messaging module.
+ */
+package org.springframework.jms.messaging;

spring-jms/src/main/java/org/springframework/jms/support/converter/MessagingMessageConverter.java

@@ -0,0 +1,116 @@
+/*
+ * Copyright 2002-2014 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.jms.support.converter;
+
+import java.util.Map;
+
+import javax.jms.JMSException;
+import javax.jms.Session;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.Assert;
+
+/**
+ * Convert a {@link Message} from the messaging abstraction to
+ * and from a {@link javax.jms.Message} using an underlying
+ * {@link MessageConverter} for the payload and a {@link JmsHeaderMapper}
+ * to map the JMS headers to and from standard message headers.
+ *
+ * @author Stephane Nicoll
+ * @since 4.1
+ */
+public class MessagingMessageConverter implements MessageConverter, InitializingBean {
+
+	private MessageConverter payloadConverter;
+
+	private JmsHeaderMapper headerMapper;
+
+	/**
+	 * Create an instance with a default payload converter.
+	 * @see org.springframework.jms.support.converter.SimpleMessageConverter
+	 * @see org.springframework.jms.support.converter.SimpleJmsHeaderMapper
+	 */
+	public MessagingMessageConverter() {
+		this(new SimpleMessageConverter(), new SimpleJmsHeaderMapper());
+	}
+
+	/**
+	 * Create an instance with the specified payload converter and
+	 * header mapper.
+	 */
+	public MessagingMessageConverter(MessageConverter payloadConverter, JmsHeaderMapper headerMapper) {
+		Assert.notNull(payloadConverter, "PayloadConverter must not be null");
+		Assert.notNull(headerMapper, "HeaderMapper must not be null");
+		this.payloadConverter = payloadConverter;
+		this.headerMapper = headerMapper;
+	}
+
+	/**
+	 * Set the {@link MessageConverter} to use to convert the payload.
+	 */
+	public void setPayloadConverter(MessageConverter payloadConverter) {
+		this.payloadConverter = payloadConverter;
+	}
+
+	/**
+	 * Set the {@link JmsHeaderMapper} to use to map JMS headers to and from
+	 * standard message headers.
+	 */
+	public void setHeaderMapper(JmsHeaderMapper headerMapper) {
+		this.headerMapper = headerMapper;
+	}
+
+	@Override
+	public void afterPropertiesSet() {
+		Assert.notNull(this.payloadConverter, "Property 'payloadConverter' is required");
+		Assert.notNull(this.headerMapper, "Property 'headerMapper' is required");
+	}
+
+	@Override
+	public javax.jms.Message toMessage(Object object, Session session) throws JMSException, MessageConversionException {
+		if (!(object instanceof Message)) {
+			throw new IllegalArgumentException("Could not convert [" + object + "] only ["
+					+ Message.class.getName() + "] is handled by this converter");
+		}
+
+		Message<?> input = (Message<?>) object;
+		javax.jms.Message reply = this.payloadConverter.toMessage(input.getPayload(), session);
+		this.headerMapper.fromHeaders(input.getHeaders(), reply);
+		return reply;
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public Object fromMessage(javax.jms.Message message) throws JMSException, MessageConversionException {
+		Map<String, Object> mappedHeaders = this.headerMapper.toHeaders(message);
+		Object convertedObject = extractPayload(message);
+		MessageBuilder<Object> builder = (convertedObject instanceof org.springframework.messaging.Message) ?
+				MessageBuilder.fromMessage((org.springframework.messaging.Message<Object>) convertedObject) :
+				MessageBuilder.withPayload(convertedObject);
+		return builder.copyHeadersIfAbsent(mappedHeaders).build();
+	}
+
+	/**
+	 * Extract the payload of the specified {@link javax.jms.Message}
+	 */
+	protected Object extractPayload(javax.jms.Message message) throws JMSException {
+		return this.payloadConverter.fromMessage(message);
+	}
+
+}