RabbitMQ/Spring migration - Step 3: extraction of the exchange name from the event

1 MailChimpSynchronization.java

// the exchange is not mentioned anymore here
@RabbitEventListener(queue = "email.updated.mailchimp.queue")
public void emailUpdated(EmailUpdated event) {
    updateMailChimpMember(event.getOldEmail(), event.getNewEmail());
}

2 Event.java

public interface Event {

    String EXCHANGE_FIELD_NAME = "EXCHANGE_NAME";

    public static String getExchangeName(Event event) {
        return getExchangeName(event.getClass());
    }

    public static String getExchangeName(Class<? extends Event> eventClass) {
        return (String) eventClass.getDeclaredField(EXCHANGE_FIELD_NAME).get(null);
        // skipped: error handling and validation
    }
}

3 EventsService.java

// just a single fire method defined once and for all, no more overloads
public void fire(Event event) {
    String exchangeName = Event.getExchangeName(event);
    String noRoutingKey = "";
    rabbitTemplate.convertAndSend(exchangeName, noRoutingKey, event);
}

4 RabbitConfig.java

// No more @PostConstruct configuration for individual exchanges, queues or bindings!
// Only generic plumbing is left.

// additionally, we scan the classpath for Event classes to find all the exchanges to create
@PostConstruct
public void createExchangesFromEventsInClasspath() {
    ClassPathScanningCandidateComponentProvider scanner = new ClassPathScanningCandidateComponentProvider(false);
    scanner.addIncludeFilter(new AssignableTypeFilter(Event.class));
    scanner.findCandidateComponents("com.hopwork").forEach(c -> {
        Class<? extends Event> eventClass = (Class<? extends Event>) Class.forName(c.getBeanClassName());
        String exchangeName = Event.getExchangeName(eventClass);

        // this method has been defined during previous step, we just made it visible and static,
        // and gave it beanRegistry as a parameter
        RabbitEventListenerPostProcessor.ensureExchangeExists(exchangeName, beanRegistry);
    });
}

5 RabbitEventListenerPostProcessor.java

// this method changes a bit to use Event.getExchangeName()
private void initRabbitListener(Object bean, String beanName, Method method) {
    // same code as in previous step
    RabbitEventListener bindingDescriptor = AnnotationUtils.findAnnotation(method, RabbitEventListener.class);
    if (bindingDescriptor != null) {
        // Previously, was:
        // - FanoutExchange exchange = ensureExchangeExists(bindingDescriptor.exchange());
        
        // Now:
        Class<? extends Event> eventClass = getEventClass(method);
        String exchangeName = Event.getExchangeName(eventClass);
        FanoutExchange exchange = ensureExchangeExists(exchangeName);
        
        // same code as in previous step
        createAndBindQueueToExchangeAndDeadLetter(exchange, bindingDescriptor.queue());
        processAmqpListener(toSpringRabbitListenerAnnotation(bindingDescriptor), method, bean, beanName);
    }
}

private Class<? extends Event> getEventClass(Method method) {
    return stream(method.getParameterTypes())
            .filter(Event.class::isAssignableFrom)
            .map(cls -> (Class<Event>) cls)
            .findFirst()
            .orElseThrow(() -> new IllegalStateException(String.format(
                    "Method %s is annotated with %s but has no argument of type %s",
                    method.getName(), RabbitEventListener.class.getSimpleName(), Event.class.getSimpleName())));
}