GH-1352: Fix more Sonar issues

Author: garyrussell

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamListenerContainer.java

@@ -25,10 +25,12 @@
 import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.lang.Nullable;
+import org.springframework.rabbit.stream.support.StreamMessageProperties;
 import org.springframework.rabbit.stream.support.converter.DefaultStreamMessageConverter;
 import org.springframework.rabbit.stream.support.converter.StreamMessageConverter;
 import org.springframework.util.Assert;
 
+import com.rabbitmq.stream.Codec;
 import com.rabbitmq.stream.Consumer;
 import com.rabbitmq.stream.ConsumerBuilder;
 import com.rabbitmq.stream.Environment;
@@ -42,18 +44,14 @@
  */
 public class StreamListenerContainer implements MessageListenerContainer, BeanNameAware {
 
-	protected Log logger = LogFactory.getLog(getClass());
-
-	private final Environment environment;
+	protected Log logger = LogFactory.getLog(getClass()); // NOSONAR
 
 	private final ConsumerBuilder builder;
 
 	private StreamMessageConverter messageConverter;
 
 	private java.util.function.Consumer<ConsumerBuilder> consumerCustomizer = c -> { };
 
-	private String stream;
-
 	private Consumer consumer;
 
 	private String listenerId;
@@ -69,17 +67,24 @@ public class StreamListenerContainer implements MessageListenerContainer, BeanNa
 	 * @param environment the environment.
 	 */
 	public StreamListenerContainer(Environment environment) {
+		this(environment, null);
+	}
+
+	/**
+	 * Construct an instance using the provided environment and codec.
+	 * @param environment the environment.
+	 * @param codec the codec used to create reply messages.
+	 */
+	public StreamListenerContainer(Environment environment, @Nullable Codec codec) {
 		Assert.notNull(environment, "'environment' cannot be null");
-		this.environment = environment;
 		this.builder = environment.consumerBuilder();
-		this.messageConverter = new DefaultStreamMessageConverter(environment);
+		this.messageConverter = new DefaultStreamMessageConverter(codec);
 	}
 
 	@Override
 	public void setQueueNames(String... queueNames) {
 		Assert.isTrue(queueNames != null && queueNames.length == 1, "Only one stream is supported");
-		this.stream = queueNames[0];
-		this.builder.stream(this.stream);
+		this.builder.stream(queueNames[0]);
 	}
 
 	/**

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamMessageProperties.java renamed to spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/support/StreamMessageProperties.java

@@ -14,9 +14,10 @@
  * limitations under the License.
  */
 
-package org.springframework.rabbit.stream.listener;
+package org.springframework.rabbit.stream.support;
 
 import org.springframework.amqp.core.MessageProperties;
+import org.springframework.lang.Nullable;
 
 import com.rabbitmq.stream.MessageHandler.Context;
 
@@ -31,23 +32,23 @@ public class StreamMessageProperties extends MessageProperties {
 
 	private static final long serialVersionUID = 1L;
 
-	private final Context context;
+	private transient Context context;
 
 	/**
 	 * Create a new instance with the provided context.
 	 * @param context the context.
 	 */
-	public StreamMessageProperties(Context context) {
+	public StreamMessageProperties(@Nullable Context context) {
 		this.context = context;
 	}
 
 	/**
 	 * Return the stream {@link Context} for the message.
 	 * @return the context.
 	 */
+	@Nullable
 	public Context getContext() {
 		return this.context;
 	}
 
-
 }

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/support/converter/DefaultStreamMessageConverter.java

@@ -17,15 +17,17 @@
 package org.springframework.rabbit.stream.support.converter;
 
 import java.util.Map;
+import java.util.function.Supplier;
 
 import org.springframework.amqp.core.Message;
-import org.springframework.amqp.core.MessageBuilder;
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.support.converter.MessageConversionException;
 import org.springframework.amqp.utils.JavaUtils;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
-import com.rabbitmq.stream.Environment;
+import com.rabbitmq.stream.Codec;
+import com.rabbitmq.stream.MessageBuilder;
 import com.rabbitmq.stream.MessageBuilder.PropertiesBuilder;
 import com.rabbitmq.stream.Properties;
 import com.rabbitmq.stream.codec.WrapperMessageBuilder;
@@ -39,29 +41,33 @@
  */
 public class DefaultStreamMessageConverter implements StreamMessageConverter {
 
-	private final Environment environment;
+	private final Supplier<MessageBuilder> builder;
+
+	public DefaultStreamMessageConverter() {
+		this.builder = () -> new WrapperMessageBuilder();
+	}
 
 	/**
-	 * Construct an instance using the provided environment.
-	 * @param environment the environment.
+	 * Construct an instance using the provided codec.
+	 * @param codec the codec.
 	 */
-	public DefaultStreamMessageConverter(Environment environment) {
-		Assert.notNull(environment, "'environment' cannot be null");
-		this.environment = environment;
+	public DefaultStreamMessageConverter(@Nullable Codec codec) {
+		this.builder = () -> codec.messageBuilder();
 	}
 
 	@Override
 	public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
 		Assert.isInstanceOf(com.rabbitmq.stream.Message.class, object);
 		com.rabbitmq.stream.Message streamMessage = (com.rabbitmq.stream.Message) object;
 		toMessageProperties(streamMessage, messageProperties);
-		return MessageBuilder.withBody(streamMessage.getBodyAsBinary()).andProperties(messageProperties).build();
+		return org.springframework.amqp.core.MessageBuilder.withBody(streamMessage.getBodyAsBinary())
+				.andProperties(messageProperties)
+				.build();
 	}
 
 	@Override
 	public com.rabbitmq.stream.Message fromMessage(Message message) throws MessageConversionException {
-		// TODO get the builder from the environment's codec
-		WrapperMessageBuilder builder = new WrapperMessageBuilder();
+		MessageBuilder builder = this.builder.get();
 		PropertiesBuilder propsBuilder = builder.properties();
 		MessageProperties mProps = message.getMessageProperties();
 		JavaUtils.INSTANCE

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java

@@ -362,7 +362,6 @@ public C createListenerContainer(RabbitListenerEndpoint endpoint) {
 			.acceptIfNotNull(this.transactionManager, instance::setTransactionManager)
 			.acceptIfNotNull(this.prefetchCount, instance::setPrefetchCount)
 			.acceptIfNotNull(this.globalQos, instance::setGlobalQos)
-			.acceptIfNotNull(this.defaultRequeueRejected, instance::setDefaultRequeueRejected)
 			.acceptIfNotNull(this.adviceChain, instance::setAdviceChain)
 			.acceptIfNotNull(this.recoveryBackOff, instance::setRecoveryBackOff)
 			.acceptIfNotNull(this.mismatchedQueuesFatal, instance::setMismatchedQueuesFatal)

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/BaseRabbitListenerContainerFactory.java

@@ -19,7 +19,6 @@
 import java.util.Arrays;
 
 import org.springframework.amqp.core.MessagePostProcessor;
-import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
 import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
@@ -42,7 +41,7 @@
 public abstract class BaseRabbitListenerContainerFactory<C extends MessageListenerContainer>
 	implements RabbitListenerContainerFactory<C> {
 
-	protected Boolean defaultRequeueRejected;
+	private Boolean defaultRequeueRejected;
 
 	private MessagePostProcessor[] beforeSendReplyPostProcessors;
 
@@ -55,7 +54,7 @@ public abstract class BaseRabbitListenerContainerFactory<C extends MessageListen
 
 	/**
 	 * @param requeueRejected true to reject by default.
-	 * @see AbstractMessageListenerContainer#setDefaultRequeueRejected
+	 * @see org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer#setDefaultRequeueRejected
 	 */
 	public void setDefaultRequeueRejected(Boolean requeueRejected) {
 		this.defaultRequeueRejected = requeueRejected;
@@ -109,7 +108,7 @@ protected void applyCommonOverrides(@Nullable RabbitListenerEndpoint endpoint, C
 		if (instance.getMessageListener() instanceof AbstractAdaptableMessageListener) {
 			AbstractAdaptableMessageListener messageListener = (AbstractAdaptableMessageListener) instance
 					.getMessageListener();
-			JavaUtils.INSTANCE
+			JavaUtils.INSTANCE // NOSONAR
 					.acceptIfNotNull(this.beforeSendReplyPostProcessors,
 							messageListener::setBeforeSendReplyPostProcessors)
 					.acceptIfNotNull(this.retryTemplate, messageListener::setRetryTemplate)

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/HandlerAdapter.java

@@ -64,7 +64,7 @@ public HandlerAdapter(DelegatingInvocableHandler delegatingHandler) {
 	 * @throws Exception if one occurs.
 	 */
 	public InvocationResult invoke(@Nullable Message<?> message, Object... providedArgs) throws Exception { // NOSONAR
-		if (this.invokerHandlerMethod != null) {
+		if (this.invokerHandlerMethod != null) { // NOSONAR (nullable message)
 			return new InvocationResult(this.invokerHandlerMethod.invoke(message, providedArgs),
 					null, this.invokerHandlerMethod.getMethod().getGenericReturnType(),
 					this.invokerHandlerMethod.getBean(),