AMQP-799: Add default @RabbitHandler support

JIRA: https://jira.spring.io/browse/AMQP-799

Support the designation of a single `@RabbitHandler` as the default if there is no
match.

Author: garyrussell

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -48,4 +48,12 @@
 @Documented
 public @interface RabbitHandler {
 
+	/**
+	 * When true, designate that this is the default fallback method if the payload type
+	 * matches no other {@link RabbitHandler} method. Only one method can be so designated.
+	 * @return true if this is the default method.
+	 * @since 2.0.3
+	 */
+	boolean isDefault() default false;
+
 }

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2017 the original author or authors.
+ * Copyright 2014-2018 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.
@@ -342,11 +342,20 @@ private Collection<RabbitListener> findListenerAnnotations(Method method) {
 	private void processMultiMethodListeners(RabbitListener[] classLevelListeners, Method[] multiMethods,
 			Object bean, String beanName) {
 		List<Method> checkedMethods = new ArrayList<Method>();
+		Method defaultMethod = null;
 		for (Method method : multiMethods) {
-			checkedMethods.add(checkProxy(method, bean));
+			Method checked = checkProxy(method, bean);
+			if (AnnotationUtils.findAnnotation(method, RabbitHandler.class).isDefault()) {
+				final Method toAssert = defaultMethod;
+				Assert.state(toAssert == null, () -> "Only one @RabbitHandler can be marked 'isDefault', found: "
+						+ toAssert.toString() + " and " + method.toString());
+				defaultMethod = checked;
+			}
+			checkedMethods.add(checked);
 		}
 		for (RabbitListener classLevelListener : classLevelListeners) {
-			MultiMethodRabbitListenerEndpoint endpoint = new MultiMethodRabbitListenerEndpoint(checkedMethods, bean);
+			MultiMethodRabbitListenerEndpoint endpoint =
+					new MultiMethodRabbitListenerEndpoint(checkedMethods, defaultMethod, bean);
 			endpoint.setBeanFactory(this.beanFactory);
 			processListener(endpoint, classLevelListener, bean, bean.getClass(), beanName);
 		}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2016 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -23,6 +23,7 @@
 import org.springframework.amqp.rabbit.listener.adapter.DelegatingInvocableHandler;
 import org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter;
 import org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
 
 /**
@@ -34,22 +35,46 @@ public class MultiMethodRabbitListenerEndpoint extends MethodRabbitListenerEndpo
 
 	private final List<Method> methods;
 
+	private final Method defaultMethod;
+
 	private DelegatingInvocableHandler delegatingHandler;
 
+	/**
+	 * Construct an instance for the provided methods and bean.
+	 * @param methods the methods.
+	 * @param bean the bean.
+	 */
 	public MultiMethodRabbitListenerEndpoint(List<Method> methods, Object bean) {
+		this(methods, null, bean);
+	}
+
+	/**
+	 * Construct an instance for the provided methods, default method and bean.
+	 * @param methods the methods.
+	 * @param defaultMethod the default method.
+	 * @param bean the bean.
+	 * @since 2.0.3
+	 */
+	public MultiMethodRabbitListenerEndpoint(List<Method> methods, @Nullable Method defaultMethod, Object bean) {
 		this.methods = methods;
+		this.defaultMethod = defaultMethod;
 		setBean(bean);
 	}
 
 	@Override
 	protected HandlerAdapter configureListenerAdapter(MessagingMessageListenerAdapter messageListener) {
 		List<InvocableHandlerMethod> invocableHandlerMethods = new ArrayList<InvocableHandlerMethod>();
+		InvocableHandlerMethod defaultHandler = null;
 		for (Method method : this.methods) {
-			invocableHandlerMethods.add(getMessageHandlerMethodFactory()
-					.createInvocableHandlerMethod(getBean(), method));
+			InvocableHandlerMethod handler = getMessageHandlerMethodFactory()
+					.createInvocableHandlerMethod(getBean(), method);
+			invocableHandlerMethods.add(handler);
+			if (method.equals(this.defaultMethod)) {
+				defaultHandler = handler;
+			}
 		}
-		this.delegatingHandler = new DelegatingInvocableHandler(invocableHandlerMethods, getBean(), getResolver(),
-				getBeanExpressionContext());
+		this.delegatingHandler = new DelegatingInvocableHandler(invocableHandlerMethods, defaultHandler,
+				getBean(), getResolver(), getBeanExpressionContext());
 		return new HandlerAdapter(this.delegatingHandler);
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -36,6 +36,7 @@
 import org.springframework.expression.ParserContext;
 import org.springframework.expression.common.TemplateParserContext;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.messaging.handler.annotation.Payload;
@@ -64,6 +65,8 @@ public class DelegatingInvocableHandler {
 	private final ConcurrentMap<Class<?>, InvocableHandlerMethod> cachedHandlers =
 			new ConcurrentHashMap<Class<?>, InvocableHandlerMethod>();
 
+	private final InvocableHandlerMethod defaultHandler;
+
 	private final Map<InvocableHandlerMethod, Expression> handlerSendTo =
 			new HashMap<InvocableHandlerMethod, Expression>();
 
@@ -82,7 +85,23 @@ public class DelegatingInvocableHandler {
 	 */
 	public DelegatingInvocableHandler(List<InvocableHandlerMethod> handlers, Object bean,
 			BeanExpressionResolver beanExpressionResolver, BeanExpressionContext beanExpressionContext) {
+		this(handlers, null, bean, beanExpressionResolver, beanExpressionContext);
+	}
+
+	/**
+	 * Construct an instance with the supplied handlers for the bean.
+	 * @param handlers the handlers.
+	 * @param defaultHandler the default handler.
+	 * @param bean the bean.
+	 * @param beanExpressionResolver the resolver.
+	 * @param beanExpressionContext the context.
+	 * @since 2.0.3
+	 */
+	public DelegatingInvocableHandler(List<InvocableHandlerMethod> handlers,
+			@Nullable InvocableHandlerMethod defaultHandler, Object bean, BeanExpressionResolver beanExpressionResolver,
+			BeanExpressionContext beanExpressionContext) {
 		this.handlers = new ArrayList<InvocableHandlerMethod>(handlers);
+		this.defaultHandler = defaultHandler;
 		this.bean = bean;
 		this.resolver = beanExpressionResolver;
 		this.beanExpressionContext = beanExpressionContext;
@@ -178,13 +197,19 @@ protected InvocableHandlerMethod findHandlerForPayload(Class<? extends Object> p
 		for (InvocableHandlerMethod handler : this.handlers) {
 			if (matchHandlerMethod(payloadClass, handler)) {
 				if (result != null) {
-					throw new AmqpException("Ambiguous methods for payload type: " + payloadClass + ": " +
-							result.getMethod().getName() + " and " + handler.getMethod().getName());
+					boolean resultIsDefault = result.equals(this.defaultHandler);
+					if (!handler.equals(this.defaultHandler) && !resultIsDefault) {
+						throw new AmqpException("Ambiguous methods for payload type: " + payloadClass + ": " +
+								result.getMethod().getName() + " and " + handler.getMethod().getName());
+					}
+					if (!resultIsDefault) {
+						continue; // otherwise replace the result with the actual match
+					}
 				}
 				result = handler;
 			}
 		}
-		return result;
+		return result != null ? result : this.defaultHandler;
 	}
 
 	protected boolean matchHandlerMethod(Class<? extends Object> payloadClass, InvocableHandlerMethod handler) {
@@ -237,4 +262,8 @@ public Method getMethodFor(Object payload) {
 		return handlerForPayload == null ? null : handlerForPayload.getMethod();
 	}
 
+	public boolean hasDefaultHandler() {
+		return this.defaultHandler != null;
+	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -50,6 +50,13 @@ public Object invoke(Message<?> message, Object... providedArgs) throws Exceptio
 		if (this.invokerHandlerMethod != null) {
 			return this.invokerHandlerMethod.invoke(message, providedArgs);
 		}
+		else if (this.delegatingHandler.hasDefaultHandler()) {
+			// Needed to avoid returning raw Message which matches Object
+			Object[] args = new Object[providedArgs.length + 1];
+			args[0] = message.getPayload();
+			System.arraycopy(providedArgs, 0, args, 1, providedArgs.length);
+			return this.delegatingHandler.invoke(message, args);
+		}
 		else {
 			return this.delegatingHandler.invoke(message, providedArgs);
 		}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2017 the original author or authors.
+ * Copyright 2014-2018 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.
@@ -318,6 +318,10 @@ public void commas() {
 
 	@Test
 	public void multiListener() {
+		Foo foo = new Foo();
+		foo.field = "foo";
+		assertEquals("FOO: foo handled by default handler",
+				rabbitTemplate.convertSendAndReceive("multi.exch", "multi.rk", foo));
 		Bar bar = new Bar();
 		bar.field = "bar";
 		rabbitTemplate.convertAndSend("multi.exch", "multi.rk", bar);
@@ -1340,6 +1344,14 @@ public String qux(@Header("amqp_receivedRoutingKey") String rk, @NonNull @Payloa
 			return "QUX: " + qux.field + ": " + rk;
 		}
 
+		@RabbitHandler(isDefault = true)
+		public String defaultHandler(@Payload Object payload) {
+			if (payload instanceof Foo) {
+				return "FOO: " + ((Foo) payload).field + " handled by default handler";
+			}
+			return payload.toString() + " handled by default handler";
+		}
+
 	}
 
 	@RabbitListener(id = "multi", bindings = @QueueBinding