GH-1455: AdviceChain on Stream Listener Container

Resolves #1455

Add an advice chain to the stream listener container and its factory.
Add a `StreamMessageRecoverer` for native stream messages.
Add a retry interceptor to work with native stream messages.

**cherry-pick to 2.4.x**

* Add since to new setter.

Author: garyrussell

Diff for: spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/config/StreamRabbitListenerContainerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-2022 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.
@@ -18,12 +18,15 @@
 
 import java.lang.reflect.Method;
 
+import org.aopalliance.aop.Advice;
+
 import org.springframework.amqp.rabbit.batch.BatchingStrategy;
 import org.springframework.amqp.rabbit.config.BaseRabbitListenerContainerFactory;
 import org.springframework.amqp.rabbit.config.ContainerCustomizer;
 import org.springframework.amqp.rabbit.listener.MethodRabbitListenerEndpoint;
 import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
 import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
+import org.springframework.amqp.utils.JavaUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.rabbit.stream.listener.ConsumerCustomizer;
 import org.springframework.rabbit.stream.listener.StreamListenerContainer;
@@ -96,9 +99,10 @@ public StreamListenerContainer createListenerContainer(RabbitListenerEndpoint en
 							});
 		}
 		StreamListenerContainer container = createContainerInstance();
-		if (this.consumerCustomizer != null) {
-			container.setConsumerCustomizer(this.consumerCustomizer);
-		}
+		Advice[] adviceChain = getAdviceChain();
+		JavaUtils.INSTANCE
+				.acceptIfNotNull(this.consumerCustomizer, container::setConsumerCustomizer)
+				.acceptIfNotNull(adviceChain, container::setAdviceChain);
 		applyCommonOverrides(endpoint, container);
 		if (this.containerCustomizer != null) {
 			this.containerCustomizer.configure(container);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-2022 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.
@@ -16,13 +16,16 @@
 
 package org.springframework.rabbit.stream.listener;
 
+import org.aopalliance.aop.Advice;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageListener;
 import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.aop.support.DefaultPointcutAdvisor;
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.lang.Nullable;
 import org.springframework.rabbit.stream.support.StreamMessageProperties;
@@ -62,6 +65,10 @@ public class StreamListenerContainer implements MessageListenerContainer, BeanNa
 
 	private MessageListener messageListener;
 
+	private StreamMessageListener streamListener;
+
+	private Advice[] adviceChain;
+
 	/**
 	 * Construct an instance using the provided environment.
 	 * @param environment the environment.
@@ -154,6 +161,18 @@ public void setAutoStartup(boolean autoStart) {
 	public boolean isAutoStartup() {
 		return this.autoStartup;
 	}
+
+	/**
+	 * Set an advice chain to apply to the listener.
+	 * @param advices the advice chain.
+	 * @since 2.4.5
+	 */
+	public void setAdviceChain(Advice... advices) {
+		Assert.notNull(advices, "'advices' cannot be null");
+		Assert.noNullElements(advices, "'advices' cannot have null elements");
+		this.adviceChain = advices;
+	}
+
 	@Override
 	@Nullable
 	public Object getMessageListener() {
@@ -183,26 +202,46 @@ public synchronized void stop() {
 
 	@Override
 	public void setupMessageListener(MessageListener messageListener) {
-		this.messageListener = messageListener;
+		adviseIfNeeded(messageListener);
 		this.builder.messageHandler((context, message) -> {
-			if (messageListener instanceof StreamMessageListener) {
-				((StreamMessageListener) messageListener).onStreamMessage(message, context);
+			if (this.streamListener != null) {
+				this.streamListener.onStreamMessage(message, context);
 			}
 			else {
 				Message message2 = this.streamConverter.toMessage(message, new StreamMessageProperties(context));
-				if (messageListener instanceof ChannelAwareMessageListener) {
+				if (this.messageListener instanceof ChannelAwareMessageListener) {
 					try {
-						((ChannelAwareMessageListener) messageListener).onMessage(message2, null);
+						((ChannelAwareMessageListener) this.messageListener).onMessage(message2, null);
 					}
 					catch (Exception e) { // NOSONAR
 						this.logger.error("Listner threw an exception", e);
 					}
 				}
 				else {
-					messageListener.onMessage(message2);
+					this.messageListener.onMessage(message2);
 				}
 			}
 		});
 	}
 
+	private void adviseIfNeeded(MessageListener messageListener) {
+		this.messageListener = messageListener;
+		if (messageListener instanceof StreamMessageListener) {
+			this.streamListener = (StreamMessageListener) messageListener;
+		}
+		if (this.adviceChain != null && this.adviceChain.length > 0) {
+			ProxyFactory factory = new ProxyFactory(messageListener);
+			for (Advice advice : this.adviceChain) {
+				factory.addAdvisor(new DefaultPointcutAdvisor(advice));
+			}
+			factory.setInterfaces(messageListener.getClass().getInterfaces());
+			if (this.streamListener != null) {
+				this.streamListener = (StreamMessageListener) factory.getProxy(getClass().getClassLoader());
+			}
+			else {
+				this.messageListener = (MessageListener) factory.getProxy(getClass().getClassLoader());
+			}
+		}
+	}
+
 }

Diff for: spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/adapter/StreamMessageListenerAdapter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-2022 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.
@@ -21,6 +21,7 @@
 import org.springframework.amqp.rabbit.listener.adapter.InvocationResult;
 import org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter;
 import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
+import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException;
 import org.springframework.rabbit.stream.listener.StreamMessageListener;
 
 import com.rabbitmq.stream.Message;
@@ -60,7 +61,7 @@ public void onStreamMessage(Message message, Context context) {
 			}
 		}
 		catch (Exception ex) {
-			this.logger.error("Failed to invoke listener", ex);
+			throw new ListenerExecutionFailedException("Failed to invoke listener", ex);
 		}
 	}
 

Diff for: spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/retry/StreamMessageRecoverer.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2022 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
+ *
+ *      https://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.rabbit.stream.retry;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.retry.MessageRecoverer;
+
+import com.rabbitmq.stream.MessageHandler.Context;
+
+/**
+ * Implementations of this interface can handle failed messages after retries are
+ * exhausted.
+ *
+ * @author Gary Russell
+ * @since 2.4.5
+ *
+ */
+@FunctionalInterface
+public interface StreamMessageRecoverer extends MessageRecoverer {
+
+	@Override
+	default void recover(Message message, Throwable cause) {
+	}
+
+	/**
+	 * Callback for message that was consumed but failed all retry attempts.
+	 *
+	 * @param message the message to recover.
+	 * @param context the context.
+	 * @param cause the cause of the error.
+	 */
+	void recover(com.rabbitmq.stream.Message message, Context context, Throwable cause);
+
+}

Diff for: spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/retry/StreamRetryOperationsInterceptorFactoryBean.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2022 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
+ *
+ *      https://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.rabbit.stream.retry;
+
+import org.springframework.amqp.rabbit.config.StatelessRetryOperationsInterceptorFactoryBean;
+import org.springframework.amqp.rabbit.retry.MessageRecoverer;
+import org.springframework.rabbit.stream.listener.StreamListenerContainer;
+import org.springframework.retry.RetryOperations;
+import org.springframework.retry.interceptor.MethodInvocationRecoverer;
+import org.springframework.retry.support.RetryTemplate;
+
+import com.rabbitmq.stream.Message;
+import com.rabbitmq.stream.MessageHandler.Context;
+
+/**
+ * Convenient factory bean for creating a stateless retry interceptor for use in a
+ * {@link StreamListenerContainer} when consuming native stream messages, giving you a
+ * large amount of control over the behavior of a container when a listener fails. To
+ * control the number of retry attempt or the backoff in between attempts, supply a
+ * customized {@link RetryTemplate}. Stateless retry is appropriate if your listener can
+ * be called repeatedly between failures with no side effects. The semantics of stateless
+ * retry mean that a listener exception is not propagated to the container until the retry
+ * attempts are exhausted. When the retry attempts are exhausted it can be processed using
+ * a {@link StreamMessageRecoverer} if one is provided.
+ *
+ * @author Gary Russell
+ *
+ * @see RetryOperations#execute(org.springframework.retry.RetryCallback,org.springframework.retry.RecoveryCallback)
+ */
+public class StreamRetryOperationsInterceptorFactoryBean extends StatelessRetryOperationsInterceptorFactoryBean {
+
+	@Override
+	protected MethodInvocationRecoverer<?> createRecoverer() {
+		return (args, cause) -> {
+			StreamMessageRecoverer messageRecoverer = (StreamMessageRecoverer) getMessageRecoverer();
+			Object arg = args[0];
+			if (arg instanceof org.springframework.amqp.core.Message) {
+				return super.recover(args, cause);
+			}
+			else {
+				if (messageRecoverer == null) {
+					this.logger.warn("Message(s) dropped on recovery: " + arg, cause);
+				}
+				else {
+					messageRecoverer.recover((Message) arg, (Context) args[1], cause);
+				}
+				return null;
+			}
+		};
+	}
+
+	/**
+	 * Set a {@link StreamMessageRecoverer} to call when retries are exhausted.
+	 * @param messageRecoverer the recoverer.
+	 */
+	public void setStreamMessageRecoverer(StreamMessageRecoverer messageRecoverer) {
+		super.setMessageRecoverer(messageRecoverer);
+	}
+
+	@Override
+	public void setMessageRecoverer(MessageRecoverer messageRecoverer) {
+		throw new UnsupportedOperationException("Use setStreamMessageRecoverer() instead");
+	}
+
+}

Diff for: spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/retry/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides classes supporting retries.
+ */
+package org.springframework.rabbit.stream.retry;