refactor: refactor rabbitmq infrastructure

Author: meysamhadeli

booking.rest

@@ -1,5 +1,5 @@
 
-@api-gateway=http://localhost:8081  
+@api-gateway=http://localhost:8081
 @keycloak-api=http://localhost:8080
 @flight-api=http://localhost:8082
 @passenger-api=http://localhost:8083
@@ -25,9 +25,9 @@ POST {{keycloak-api}}/realms/keycloak-realm/protocol/openid-connect/token
 Content-Type: application/x-www-form-urlencoded
 
 grant_type=client_credentials
-&client_id=flight-client-credentials
-&client_secret=Glcz8z8E6alTIfay1GdMc6XTTeuZrgOs
-&scope=openid
+  &client_id=flight-client-credentials
+  &client_secret=Glcz8z8E6alTIfay1GdMc6XTTeuZrgOs
+  &scope=openid
 ###
 
 
@@ -97,13 +97,13 @@ authorization: bearer {{Authenticate.response.body.access_token}}
 
 ###
 # @name Create_Flights
-POST {{flight-api}}/api/v1/flight
+POST {{api-gateway}}/api/v1/flight
 accept: application/json
 Content-Type: application/json
 authorization: bearer {{Authenticate.response.body.access_token}}
 
 {
-  "flightNumber": "3dd5555655533",
+  "flightNumber": "20BB",
   "aircraftId": {{aircraftId}},
   "departureAirportId": {{departureAirportId}},
   "departureDate": "2022-03-01T14:55:41.255Z",
@@ -119,13 +119,13 @@ authorization: bearer {{Authenticate.response.body.access_token}}
 
 ###
 # @name Update_Flights
-PUT  {{flight-api}}/api/v1/flight/019498d8-2a28-7f0b-ac0c-2f8352c13c79
+PUT  {{api-gateway}}/api/v1/flight/01949849-1608-7e16-975a-e7f4cf1d029d
 accept: application/json
 Content-Type: application/json
 authorization: bearer {{Authenticate.response.body.access_token}}
 
 {
-  "flightNumber": "2222222222222222",
+  "flightNumber": "20BB",
   "aircraftId": {{aircraftId}},
   "departureAirportId": {{departureAirportId}},
   "departureDate": "2025-01-24T12:35:11.803Z",
@@ -134,7 +134,7 @@ authorization: bearer {{Authenticate.response.body.access_token}}
   "durationMinutes": 140,
   "flightDate": "2025-01-24T12:35:11.803Z",
   "status": "Flying",
-  "price": 9000,
+  "price": 8000,
   "isDeleted": false
 }
 ###

src/buildingblocks/src/main/java/buildingblocks/core/event/EventDispatcherConfiguration.java

@@ -5,6 +5,7 @@
 import org.hibernate.engine.spi.PersistenceContext;
 import org.hibernate.event.spi.PersistContext;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingClass;
+import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
@@ -13,7 +14,7 @@ public class EventDispatcherConfiguration {
 
     @Bean
     @ConditionalOnMissingClass
-    public EventDispatcher eventDispatcher(EventMapper eventMapper, PersistMessageProcessor persistMessageProcessor) {
-        return new EventDispatcherImpl(eventMapper, persistMessageProcessor);
+    public EventDispatcher eventDispatcher(EventMapper eventMapper, PersistMessageProcessor persistMessageProcessor, ApplicationContext applicationContext) {
+        return new EventDispatcherImpl(eventMapper, persistMessageProcessor, applicationContext);
     }
 }

src/buildingblocks/src/main/java/buildingblocks/core/event/EventDispatcherImpl.java

@@ -7,6 +7,7 @@
 import org.reflections.Reflections;
 import org.reflections.scanners.SubTypesScanner;
 import org.reflections.util.ConfigurationBuilder;
+import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Import;
 import org.springframework.stereotype.Component;
 
@@ -22,11 +23,13 @@
 public class EventDispatcherImpl implements EventDispatcher {
     private final EventMapper eventMapper;
     private final PersistMessageProcessor persistMessageProcessor;
+  private final ApplicationContext applicationContext;
 
-    public EventDispatcherImpl(EventMapper eventMapper, PersistMessageProcessor persistMessageProcessor) {
+  public EventDispatcherImpl(EventMapper eventMapper, PersistMessageProcessor persistMessageProcessor, ApplicationContext applicationContext) {
         this.eventMapper = eventMapper;
         this.persistMessageProcessor = persistMessageProcessor;
-    }
+    this.applicationContext = applicationContext;
+  }
 
     @Override
     public <T extends DomainEvent> void send(List<T> domainEvents, Class<?> eventType) {
@@ -42,13 +45,13 @@ public <T extends DomainEvent> void send(List<T> domainEvents, Class<?> eventTyp
 
     @Override
     public List<DomainEvent> getDomainEvents() {
-            AggregateRoot<?> aggregateRoot = ReflectionUtils.getInstanceOfSubclass(AggregateRoot.class);
+            AggregateRoot<?> aggregateRoot = ReflectionUtils.getInstanceOfSubclass(AggregateRoot.class, applicationContext);
             return Objects.requireNonNull(aggregateRoot).getDomainEvents();
     }
 
     @Override
     public void clearDomainEvents() {
-        AggregateRoot<?> aggregateRoot = ReflectionUtils.getInstanceOfSubclass(AggregateRoot.class);
+        AggregateRoot<?> aggregateRoot = ReflectionUtils.getInstanceOfSubclass(AggregateRoot.class, applicationContext);
          Objects.requireNonNull(aggregateRoot).clearDomainEvents();
     }
 }

src/buildingblocks/src/main/java/buildingblocks/outboxprocessor/PersistMessageProcessorImpl.java

@@ -152,7 +152,7 @@ private boolean processOutbox(PersistMessageEntity message) {
       if (data instanceof IntegrationEvent integrationEvent) {
 
         this.rabbitTemplate.convertSendAndReceive(
-          message.getDataType(),
+          rabbitProperties.getTemplate().getExchange(),
           message.getDataType(),
           JsonConverterUtils.serializeObject(integrationEvent), msg -> {
             MessageProperties props = msg.getMessageProperties();

src/buildingblocks/src/main/java/buildingblocks/rabbitmq/RabbitmqConfiguration.java

@@ -15,11 +15,12 @@
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
+import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.*;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.support.TransactionTemplate;
-
 import java.util.HashMap;
+import java.util.Map;
 import java.util.UUID;
 import java.util.function.Supplier;
 
@@ -42,7 +43,6 @@ public RabbitmqConfiguration(
     this.logger = logger;
   }
 
-
   @Bean
   public ConnectionFactory connectionFactory() {
     CachingConnectionFactory connectionFactory = new CachingConnectionFactory(rabbitProperties.getHost());
@@ -54,39 +54,9 @@ public ConnectionFactory connectionFactory() {
 
   @Bean
   public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
-   return new RabbitAdmin(connectionFactory);
+    return new RabbitAdmin(connectionFactory);
   }
 
-//  public String getQueueName() {
-//    return rabbitProperties.getTemplate().getExchange() + "_queue";
-//  }
-
-//  @Bean
-//  public Queue queue() {
-//    return new Queue(rabbitProperties.getTemplate().getExchange() + "_queue", true);
-//  }
-//
-//  @Bean
-//  public Exchange exchange() {
-//    return switch (exchangeType.toLowerCase()) {
-//      case "direct" -> new DirectExchange(rabbitProperties.getTemplate().getExchange());
-//      case "fanout" -> new FanoutExchange(rabbitProperties.getTemplate().getExchange());
-//      default -> new TopicExchange(rabbitProperties.getTemplate().getExchange());
-//    };
-//  }
-//
-//
-//  @Bean
-//  public Binding binding(Queue queue, Exchange exchange) {
-//    String routingKey = rabbitProperties.getTemplate().getExchange() + "_routing_key";
-//    return switch (exchange) {
-//      case TopicExchange topicExchange -> BindingBuilder.bind(queue).to(topicExchange).with(routingKey);
-//      case DirectExchange directExchange -> BindingBuilder.bind(queue).to(directExchange).with(routingKey);
-//      case FanoutExchange fanoutExchange -> BindingBuilder.bind(queue).to(fanoutExchange);
-//      case null, default -> throw new IllegalArgumentException("Unsupported exchange type for binding");
-//    };
-//  }
-
   @Bean
   public AsyncRabbitTemplate asyncTemplate(ConnectionFactory connectionFactory) {
     return new AsyncRabbitTemplate(new RabbitTemplate(connectionFactory));
@@ -98,38 +68,35 @@ public RabbitTemplate template(ConnectionFactory connectionFactory) {
   }
 
   @Bean
-  public MessageListenerContainer addListeners(PersistMessageProcessor persistMessageProcessor, RabbitAdmin rabbitAdmin) {
+  public MessageListenerContainer addListeners(
+    PersistMessageProcessor persistMessageProcessor,
+    RabbitAdmin rabbitAdmin,
+    ApplicationContext applicationContext) {
     SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
     container.setConnectionFactory(connectionFactory());
-
-//    // Change to support multiple listeners
-//    List<MessageListener> listeners = ReflectionUtils.getAllInstanceOfSubclasses(MessageListener.class);
-//
+    container.setAcknowledgeMode(AcknowledgeMode.AUTO);
 
     // Map to store message type to listener mapping
     HashMap<Class<?>, Supplier<MessageListener>> listenerMap = new HashMap<>();
 
-    // Find all MessageListener implementations
-    var listener = ReflectionUtils.getInstanceOfSubclass(MessageListener.class);
-    if (listener != null) {
+    // Retrieve all beans of type MessageListener
+    Map<String, MessageListener> listeners = applicationContext.getBeansOfType(MessageListener.class);
+
+    listeners.values().forEach(listener -> {
       RabbitmqMessageHandler annotation = listener.getClass().getAnnotation(RabbitmqMessageHandler.class);
 
       if (annotation != null) {
         var typeName = annotation.messageType().getTypeName();
 
-        DirectExchange directExchange = new DirectExchange(typeName);
-        rabbitAdmin.declareExchange(directExchange);
+        Queue queue = declareQueue(rabbitAdmin, typeName);
 
-        Queue queue = new Queue(typeName + "_queue", true);
-        rabbitAdmin.declareQueue(queue);
+        Exchange exchange = declareExchange(rabbitAdmin);
 
-        // Bind the queue to the fanout exchange
-        rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(directExchange).with(typeName));
+        declareBindings(rabbitAdmin, queue, exchange, typeName);
 
-        // Set the queue name from the annotation
         container.setQueueNames(queue.getName());
 
-        // Store a supplier for the listener
+        // Store the listener in the map
         listenerMap.put(annotation.messageType(), () -> listener);
       }
 
@@ -141,7 +108,6 @@ public MessageListenerContainer addListeners(PersistMessageProcessor persistMess
             PersistMessageEntity persistMessage = persistMessageProcessor.existInboxMessage(id);
 
             if (persistMessage == null) {
-
               Class<?> messageType = ReflectionUtils.findClassFromName(message.getMessageProperties().getType());
 
               // Find the appropriate handler
@@ -162,7 +128,37 @@ public MessageListenerContainer addListeners(PersistMessageProcessor persistMess
           return null;
         });
       });
-    }
+    });
     return container;
   }
+
+  private static void declareBindings(RabbitAdmin rabbitAdmin, Queue queue, Exchange exchange, String routingKey) {
+    switch (exchange) {
+      case TopicExchange topicExchange -> rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(topicExchange).with(routingKey));
+      case DirectExchange directExchange -> rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(directExchange).with(routingKey));
+      case FanoutExchange fanoutExchange -> rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(fanoutExchange));
+      case null, default -> throw new IllegalArgumentException("Unsupported exchange type for binding");
+    };
+  }
+
+  private Exchange declareExchange(RabbitAdmin rabbitAdmin) {
+    Exchange exchange = switch (exchangeType.toLowerCase()) {
+      case "direct" ->
+        new DirectExchange(rabbitProperties.getTemplate().getExchange());
+      case "fanout" ->
+        new FanoutExchange(rabbitProperties.getTemplate().getExchange());
+      default ->
+        new TopicExchange(rabbitProperties.getTemplate().getExchange());
+    };
+
+    rabbitAdmin.declareExchange(exchange);
+
+    return exchange;
+  }
+
+  private static Queue declareQueue(RabbitAdmin rabbitAdmin, String typeName) {
+    Queue queue = new Queue(typeName + "_queue", true);
+    rabbitAdmin.declareQueue(queue);
+    return queue;
+  }
 }

src/buildingblocks/src/main/java/buildingblocks/rabbitmq/RabbitmqMessageHandler.java

@@ -13,10 +13,4 @@
    * @return the message type class
    */
   Class<?> messageType();
-
-  /**
-   * Specifies the queue name for this message handler.
-   * @return the queue name
-   */
-  String queueName();
 }

src/buildingblocks/src/main/java/buildingblocks/utils/reflection/ReflectionUtils.java

@@ -3,17 +3,16 @@
 import org.reflections.Reflections;
 import org.reflections.scanners.SubTypesScanner;
 import org.reflections.util.ConfigurationBuilder;
+import org.springframework.context.ApplicationContext;
+
 import java.lang.reflect.Constructor;
-import java.lang.reflect.InvocationTargetException;
 import java.util.ArrayList;
-import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
 
 public final class ReflectionUtils {
-  public static <T> T getInstanceOfSubclass(Class<T> abstractClass) {
+  public static <T> T getInstanceOfSubclass(Class<T> abstractClass, ApplicationContext applicationContext) {
     try {
-      // Initialize Reflections with the provided package name
       Reflections reflections = new Reflections(new ConfigurationBuilder()
         .forPackages("")
         .setScanners(new SubTypesScanner(false))
@@ -23,49 +22,53 @@ public static <T> T getInstanceOfSubclass(Class<T> abstractClass) {
       Set<Class<? extends T>> subTypes = reflections.getSubTypesOf(abstractClass);
 
       if (!subTypes.isEmpty()) {
-        // Get the first subclass (you can enhance this to return any specific subclass if needed)
-        Class<? extends T> firstSubclass = subTypes.iterator().next();
-
-        // Create an instance of the subclass using reflection
-        Constructor<? extends T> constructor = firstSubclass.getDeclaredConstructor();
-        return constructor.newInstance();
+        for (Class<? extends T> subType : subTypes) {
+          // Check if the subclass is managed by Spring
+          if (applicationContext.containsBean(subType.getSimpleName())) {
+            // Return the Spring-managed bean
+            return applicationContext.getBean(subType);
+          } else {
+            // Fall back to creating a new instance manually
+            Constructor<? extends T> constructor = subType.getDeclaredConstructor();
+            return constructor.newInstance();
+          }
+        }
       }
       return null; // or throw an exception if no subclass is found
     } catch (Exception ex) {
       throw new RuntimeException("Error occurred while creating an instance of subclass", ex);
     }
   }
 
-  public static <T> List<T> getAllInstanceOfSubclasses(Class<T> abstractClass) {
+  public static <T> List<T> getAllInstanceOfSubclasses(Class<T> abstractClass, ApplicationContext applicationContext) {
     List<T> instances = new ArrayList<>();
     try {
-      // Initialize Reflections with the provided package name
       Reflections reflections = new Reflections(new ConfigurationBuilder()
         .forPackages("")
         .setScanners(new SubTypesScanner(false))
       );
 
-      // Get all subclasses of the abstract class
       Set<Class<? extends T>> subTypes = reflections.getSubTypesOf(abstractClass);
 
-      if (!subTypes.isEmpty()) {
-        // Get the first subclass (you can enhance this to return any specific subclass if needed)
-
-        for (Class<? extends T> subType : subTypes) {
-          try {
-            // Create a new instance using the no-argument constructor
+      for (Class<? extends T> subType : subTypes) {
+        try {
+          // Check if the class is managed by Spring
+          if (applicationContext.containsBean(subType.getSimpleName())) {
+            T instance = applicationContext.getBean(subType);
+            instances.add(instance);
+          } else {
+            // Fall back to no-argument constructor
             T instance = subType.getDeclaredConstructor().newInstance();
             instances.add(instance);
-          } catch (InstantiationException | IllegalAccessException |
-                   NoSuchMethodException | InvocationTargetException ex) {
-            throw new RuntimeException("Error occurred while creating an instance of subclasses", ex);
           }
+        } catch (Exception ex) {
+          throw new RuntimeException("Error occurred while creating an instance of subclasses", ex);
         }
       }
-      return instances;
     } catch (Exception ex) {
       throw new RuntimeException("Error occurred while creating an instance of subclass", ex);
     }
+    return instances;
   }
 
   public static Class<?> findClassFromName(String className) {