RabbitMQ/Spring migration - Step 2: let's craft our own annotation

1 MailChimpSynchronization.java

// let's use our own annotation!
@RabbitEventListener(queue = "email.updated.mailchimp.queue", exchange = RabbitConfig.EMAIL_UPDATED_EXCHANGE)
public void emailUpdated(EmailUpdated event) {
    updateMailChimpMember(event.getOldEmail(), event.getNewEmail());
}

2 RabbitEventListener.java

@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface RabbitEventListener {
    String exchange();
    String queue();
}

3 RabbitConfig.java

@PostConstruct
public void createExchanges() {
    // We don't create the queues anymore here! We don't care about the bindings either :-)
    createExchangeBean(EMAIL_UPDATED_EXCHANGE);
    // ... other exchanges...
}

@Override
@Primary
@Bean(name = RabbitListenerConfigUtils.RABBIT_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public RabbitListenerAnnotationBeanPostProcessor rabbitListenerAnnotationProcessor() {
    return new RabbitEventListenerPostProcessor(beanRegistry, backFromTheDeadExchange());
}

4 RabbitEventListenerPostProcessor.java

public class RabbitEventListenerPostProcessor extends RabbitListenerAnnotationBeanPostProcessor {

    @Override
    public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException {
        Class<?> targetClass = AopUtils.getTargetClass(bean);
        ReflectionUtils.doWithMethods(targetClass, method ->
                initRabbitListener(bean, beanName, method), ReflectionUtils.USER_DECLARED_METHODS);
        return super.postProcessAfterInitialization(bean, beanName);
    }

    private void initRabbitListener(Object bean, String beanName, Method method) {
        RabbitEventListener bindingDescriptor = AnnotationUtils.findAnnotation(method, RabbitEventListener.class);
        if (bindingDescriptor != null) {
            FanoutExchange exchange = ensureExchangeExists(bindingDescriptor.exchange());
            // same code as in previous step
            createAndBindQueueToExchangeAndDeadLetter(exchange, bindingDescriptor.queue());
            
            // now that we are sure that both the exchange and queue exist, we can let Spring do its magic
            processAmqpListener(toSpringRabbitListenerAnnotation(bindingDescriptor), method, bean, beanName);
        }
    }

    private RabbitListener toSpringRabbitListenerAnnotation(final RabbitEventListener bindingDescriptor) {
        // an annotation is just an interface, let's create an anonymous implementation!
        return new RabbitListener() {
            @Override
            public Class<? extends Annotation> annotationType() {
                return RabbitListener.class;
            }

            @Override
            public String[] queues() {
                return new String[]{bindingDescriptor.queue()};
            }
            
            // ... the other overridden methods just return the default values specified in @RabbitListener...
        };
    }

    private FanoutExchange ensureExchangeExists(String exchangeName) {
        String exchangeBeanName = beanNameForExchange(exchangeName);

        if (beanRegistry.containsBean(exchangeBeanName)) {
            return beanRegistry.getBean(exchangeBeanName, FanoutExchange.class);
        }

        FanoutExchange exchange = new FanoutExchange(exchangeName, true, false);
        beanRegistry.registerSingleton(exchangeBeanName, exchange);
        return exchange;
    }
    
    // ... more code...
}