001/* 002 * Copyright 2002-2018 the original author or authors. 003 * 004 * Licensed under the Apache License, Version 2.0 (the "License"); 005 * you may not use this file except in compliance with the License. 006 * You may obtain a copy of the License at 007 * 008 * https://www.apache.org/licenses/LICENSE-2.0 009 * 010 * Unless required by applicable law or agreed to in writing, software 011 * distributed under the License is distributed on an "AS IS" BASIS, 012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 013 * See the License for the specific language governing permissions and 014 * limitations under the License. 015 */ 016 017package org.springframework.messaging.core; 018 019import java.util.Map; 020 021import org.apache.commons.logging.Log; 022import org.apache.commons.logging.LogFactory; 023 024import org.springframework.lang.Nullable; 025import org.springframework.messaging.Message; 026import org.springframework.messaging.MessageHeaders; 027import org.springframework.messaging.MessagingException; 028import org.springframework.messaging.converter.MessageConversionException; 029import org.springframework.messaging.converter.MessageConverter; 030import org.springframework.messaging.converter.SimpleMessageConverter; 031import org.springframework.messaging.converter.SmartMessageConverter; 032import org.springframework.util.Assert; 033 034/** 035 * Abstract base class for implementations of {@link MessageSendingOperations}. 036 * 037 * @author Mark Fisher 038 * @author Rossen Stoyanchev 039 * @author Stephane Nicoll 040 * @since 4.0 041 * @param <D> the destination type 042 */ 043public abstract class AbstractMessageSendingTemplate<D> implements MessageSendingOperations<D> { 044 045 /** 046 * Name of the header that can be set to provide further information 047 * (e.g. a {@code MethodParameter} instance) about the origin of the 048 * payload, to be taken into account as a conversion hint. 049 * @since 4.2 050 */ 051 public static final String CONVERSION_HINT_HEADER = "conversionHint"; 052 053 054 protected final Log logger = LogFactory.getLog(getClass()); 055 056 @Nullable 057 private D defaultDestination; 058 059 private MessageConverter converter = new SimpleMessageConverter(); 060 061 062 /** 063 * Configure the default destination to use in send methods that don't have 064 * a destination argument. If a default destination is not configured, send methods 065 * without a destination argument will raise an exception if invoked. 066 */ 067 public void setDefaultDestination(@Nullable D defaultDestination) { 068 this.defaultDestination = defaultDestination; 069 } 070 071 /** 072 * Return the configured default destination. 073 */ 074 @Nullable 075 public D getDefaultDestination() { 076 return this.defaultDestination; 077 } 078 079 /** 080 * Set the {@link MessageConverter} to use in {@code convertAndSend} methods. 081 * <p>By default, {@link SimpleMessageConverter} is used. 082 * @param messageConverter the message converter to use 083 */ 084 public void setMessageConverter(MessageConverter messageConverter) { 085 Assert.notNull(messageConverter, "MessageConverter must not be null"); 086 this.converter = messageConverter; 087 } 088 089 /** 090 * Return the configured {@link MessageConverter}. 091 */ 092 public MessageConverter getMessageConverter() { 093 return this.converter; 094 } 095 096 097 @Override 098 public void send(Message<?> message) { 099 send(getRequiredDefaultDestination(), message); 100 } 101 102 protected final D getRequiredDefaultDestination() { 103 Assert.state(this.defaultDestination != null, "No 'defaultDestination' configured"); 104 return this.defaultDestination; 105 } 106 107 @Override 108 public void send(D destination, Message<?> message) { 109 doSend(destination, message); 110 } 111 112 protected abstract void doSend(D destination, Message<?> message); 113 114 115 @Override 116 public void convertAndSend(Object payload) throws MessagingException { 117 convertAndSend(payload, null); 118 } 119 120 @Override 121 public void convertAndSend(D destination, Object payload) throws MessagingException { 122 convertAndSend(destination, payload, (Map<String, Object>) null); 123 } 124 125 @Override 126 public void convertAndSend(D destination, Object payload, @Nullable Map<String, Object> headers) 127 throws MessagingException { 128 129 convertAndSend(destination, payload, headers, null); 130 } 131 132 @Override 133 public void convertAndSend(Object payload, @Nullable MessagePostProcessor postProcessor) 134 throws MessagingException { 135 136 convertAndSend(getRequiredDefaultDestination(), payload, postProcessor); 137 } 138 139 @Override 140 public void convertAndSend(D destination, Object payload, @Nullable MessagePostProcessor postProcessor) 141 throws MessagingException { 142 143 convertAndSend(destination, payload, null, postProcessor); 144 } 145 146 @Override 147 public void convertAndSend(D destination, Object payload, @Nullable Map<String, Object> headers, 148 @Nullable MessagePostProcessor postProcessor) throws MessagingException { 149 150 Message<?> message = doConvert(payload, headers, postProcessor); 151 send(destination, message); 152 } 153 154 /** 155 * Convert the given Object to serialized form, possibly using a 156 * {@link MessageConverter}, wrap it as a message with the given 157 * headers and apply the given post processor. 158 * @param payload the Object to use as payload 159 * @param headers the headers for the message to send 160 * @param postProcessor the post processor to apply to the message 161 * @return the converted message 162 */ 163 protected Message<?> doConvert(Object payload, @Nullable Map<String, Object> headers, 164 @Nullable MessagePostProcessor postProcessor) { 165 166 MessageHeaders messageHeaders = null; 167 Object conversionHint = (headers != null ? headers.get(CONVERSION_HINT_HEADER) : null); 168 169 Map<String, Object> headersToUse = processHeadersToSend(headers); 170 if (headersToUse != null) { 171 if (headersToUse instanceof MessageHeaders) { 172 messageHeaders = (MessageHeaders) headersToUse; 173 } 174 else { 175 messageHeaders = new MessageHeaders(headersToUse); 176 } 177 } 178 179 MessageConverter converter = getMessageConverter(); 180 Message<?> message = (converter instanceof SmartMessageConverter ? 181 ((SmartMessageConverter) converter).toMessage(payload, messageHeaders, conversionHint) : 182 converter.toMessage(payload, messageHeaders)); 183 if (message == null) { 184 String payloadType = payload.getClass().getName(); 185 Object contentType = (messageHeaders != null ? messageHeaders.get(MessageHeaders.CONTENT_TYPE) : null); 186 throw new MessageConversionException("Unable to convert payload with type='" + payloadType + 187 "', contentType='" + contentType + "', converter=[" + getMessageConverter() + "]"); 188 } 189 if (postProcessor != null) { 190 message = postProcessor.postProcessMessage(message); 191 } 192 return message; 193 } 194 195 /** 196 * Provides access to the map of input headers before a send operation. 197 * Subclasses can modify the headers and then return the same or a different map. 198 * <p>This default implementation in this class returns the input map. 199 * @param headers the headers to send (or {@code null} if none) 200 * @return the actual headers to send (or {@code null} if none) 201 */ 202 @Nullable 203 protected Map<String, Object> processHeadersToSend(@Nullable Map<String, Object> headers) { 204 return headers; 205 } 206 207}