I am trying to get my application to consume messages from rabbitmq queue but after a message has been added to the queue my application does not consume it and the logs does not show anything.
This is a spring project using the Axon framework
UserEventHandler.java
package com.jasper.ecommerce.query.handler;
import com.jasper.ecommerce.query.model.UserReadModel;
import com.jasper.ecommerce.query.query.FindAllUsersQuery;
import com.jasper.ecommerce.query.repository.UserRepository;
import com.jasper.ecommerce.shared.events.UserCreatedEvent;
import org.axonframework.eventhandling.EventHandler;
import org.axonframework.queryhandling.QueryHandler;
import org.springframework.stereotype.Component;
import java.util.List;
@Component
public class UserEventHandler {
private final UserRepository userRepository;
public UserEventHandler(UserRepository userRepository) {
this.userRepository = userRepository;
}
// Event handler for UserCreatedEvent
@EventHandler
public void on(UserCreatedEvent event) {
System.out.println("UserCreatedEvent received: " + event);
// Convert event to read model and save it in MongoDB
UserReadModel user = new UserReadModel(event.getUserId(), event.getName(), event.getEmail());
userRepository.save(user); // This inserts the document into MongoDB
}
// Query handler to return all users
@QueryHandler
public List<UserReadModel> handle(FindAllUsersQuery query) {
return userRepository.findAll();
}
}
AxonAMQPConfiguration.java
package com.jasper.ecommerce.query.config;
import org.axonframework.config.EventProcessingConfigurer;
import org.axonframework.extensions.amqp.eventhandling.DefaultAMQPMessageConverter;
import org.axonframework.extensions.amqp.eventhandling.spring.SpringAMQPMessageSource;
import org.axonframework.serialization.Serializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class AxonAMQPConfiguration {
@Bean
public SpringAMQPMessageSource rabbitMQMessageSource(Serializer serializer) {
return new SpringAMQPMessageSource(
DefaultAMQPMessageConverter.builder()
.serializer(serializer)
.build()
);
}
public void configure(EventProcessingConfigurer configurer, SpringAMQPMessageSource rabbitMQMessageSource) {
configurer.registerSubscribingEventProcessor("UserEventHandler", c -> rabbitMQMessageSource);
}
}
application.yml
spring:
application:
name: ecommerce-cqrs-query
data:
mongodb:
uri: mongodb://localhost:27017/ecommerce-query
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
axon:
amqp:
exchange: AxonExchange
eventhandling:
processors:
UserEventHandler:
mode: subscribing
axonserver:
enabled: false
distributed:
enabled: false
logging:
level:
org.axonframework: DEBUG
org.springframework.amqp: DEBUG
com.rabbitmq: DEBUG
server:
port: 8080