fix(circular-dependency): Extract listener config and remove handler resolver from replies listener

juancgalvis · juancgalvis · commit 5aa32456a93d · 2025-01-08T17:05:38.000-05:00
diff --git a/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/broker/BrokerProvider.java b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/broker/BrokerProvider.java
@@ -13,7 +13,7 @@ public interface BrokerProvider<T extends GenericAsyncProps> {
 
     DomainEventBus getDomainBus();
 
-    DirectAsyncGateway getDirectAsyncGateway(HandlerResolver resolver);
+    DirectAsyncGateway getDirectAsyncGateway();
 
     void listenDomainEvents(HandlerResolver resolver);
 
@@ -23,7 +23,7 @@ public interface BrokerProvider<T extends GenericAsyncProps> {
 
     void listenQueries(HandlerResolver resolver);
 
-    void listenReplies(HandlerResolver resolver);
+    void listenReplies();
 
     Mono<RCHealth> healthCheck();
 }
diff --git a/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsConfig.java b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsConfig.java
@@ -4,16 +4,9 @@
 import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.java.Log;
-import org.reactivecommons.async.api.DefaultCommandHandler;
-import org.reactivecommons.async.api.DefaultQueryHandler;
-import org.reactivecommons.async.api.HandlerRegistry;
-import org.reactivecommons.async.commons.HandlerResolver;
-import org.reactivecommons.async.commons.HandlerResolverBuilder;
 import org.reactivecommons.async.commons.config.BrokerConfig;
 import org.reactivecommons.async.commons.converters.json.DefaultObjectMapperSupplier;
 import org.reactivecommons.async.commons.converters.json.ObjectMapperSupplier;
-import org.reactivecommons.async.commons.ext.CustomReporter;
-import org.reactivecommons.async.commons.ext.DefaultCustomReporter;
 import org.reactivecommons.async.commons.reply.ReactiveReplyRouter;
 import org.reactivecommons.async.starter.broker.BrokerProvider;
 import org.reactivecommons.async.starter.broker.BrokerProviderFactory;
@@ -27,7 +20,6 @@
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
-import reactor.core.publisher.Mono;
 
 import java.util.Map;
 
@@ -66,24 +58,6 @@ public ConnectionManager buildConnectionManager(ApplicationContext context) {
         return connectionManager;
     }
 
-    @Bean
-    @SuppressWarnings({"rawtypes", "unchecked"})
-    public DomainHandlers buildHandlers(ApplicationContext context,
-                                        HandlerRegistry primaryRegistry, DefaultCommandHandler<?> commandHandler) {
-        DomainHandlers handlers = new DomainHandlers();
-        final Map<String, HandlerRegistry> registries = context.getBeansOfType(HandlerRegistry.class);
-        if (!registries.containsValue(primaryRegistry)) {
-            registries.put("primaryHandlerRegistry", primaryRegistry);
-        }
-        final Map<String, GenericAsyncPropsDomain> props = context.getBeansOfType(GenericAsyncPropsDomain.class);
-        props.forEach((beanName, properties) -> properties.forEach((domain, asyncProps) -> {
-            String domainName = (String) domain;
-            HandlerResolver resolver = HandlerResolverBuilder.buildResolver(domainName, registries, commandHandler);
-            handlers.add(domainName, resolver);
-        }));
-        return handlers;
-    }
-
     @Bean
     @ConditionalOnMissingBean
     public BrokerConfig brokerConfig() {
@@ -96,33 +70,6 @@ public ObjectMapperSupplier objectMapperSupplier() {
         return new DefaultObjectMapperSupplier();
     }
 
-    @Bean
-    @ConditionalOnMissingBean
-    public CustomReporter reactiveCommonsCustomErrorReporter() {
-        return new DefaultCustomReporter();
-    }
-
-    @Bean
-    @ConditionalOnMissingBean
-    @SuppressWarnings("rawtypes")
-    public DefaultQueryHandler defaultHandler() {
-        return (DefaultQueryHandler<Object, Object>) command ->
-                Mono.error(new RuntimeException("No Handler Registered"));
-    }
-
-    @Bean
-    @ConditionalOnMissingBean
-    @SuppressWarnings("rawtypes")
-    public DefaultCommandHandler defaultCommandHandler() {
-        return message -> Mono.error(new RuntimeException("No Handler Registered"));
-    }
-
-    @Bean
-    @ConditionalOnMissingBean
-    public HandlerRegistry defaultHandlerRegistry() {
-        return HandlerRegistry.register();
-    }
-
     @Bean
     @ConditionalOnMissingBean(ReactiveReplyRouter.class)
     public ReactiveReplyRouter defaultReactiveReplyRouter() {
diff --git a/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsListenersConfig.java b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/config/ReactiveCommonsListenersConfig.java
@@ -0,0 +1,70 @@
+package org.reactivecommons.async.starter.config;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.java.Log;
+import org.reactivecommons.async.api.DefaultCommandHandler;
+import org.reactivecommons.async.api.DefaultQueryHandler;
+import org.reactivecommons.async.api.HandlerRegistry;
+import org.reactivecommons.async.commons.HandlerResolver;
+import org.reactivecommons.async.commons.HandlerResolverBuilder;
+import org.reactivecommons.async.commons.ext.CustomReporter;
+import org.reactivecommons.async.commons.ext.DefaultCustomReporter;
+import org.reactivecommons.async.starter.props.GenericAsyncPropsDomain;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import reactor.core.publisher.Mono;
+
+import java.util.Map;
+
+@Log
+@Configuration
+@RequiredArgsConstructor
+public class ReactiveCommonsListenersConfig {
+
+    @Bean
+    @SuppressWarnings({"rawtypes", "unchecked"})
+    public DomainHandlers buildHandlers(ApplicationContext context,
+                                        HandlerRegistry primaryRegistry, DefaultCommandHandler<?> commandHandler) {
+        DomainHandlers handlers = new DomainHandlers();
+        final Map<String, HandlerRegistry> registries = context.getBeansOfType(HandlerRegistry.class);
+        if (!registries.containsValue(primaryRegistry)) {
+            registries.put("primaryHandlerRegistry", primaryRegistry);
+        }
+        final Map<String, GenericAsyncPropsDomain> props = context.getBeansOfType(GenericAsyncPropsDomain.class);
+        props.forEach((beanName, properties) -> properties.forEach((domain, asyncProps) -> {
+            String domainName = (String) domain;
+            HandlerResolver resolver = HandlerResolverBuilder.buildResolver(domainName, registries, commandHandler);
+            handlers.add(domainName, resolver);
+        }));
+        return handlers;
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public CustomReporter reactiveCommonsCustomErrorReporter() {
+        return new DefaultCustomReporter();
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    @SuppressWarnings("rawtypes")
+    public DefaultQueryHandler defaultHandler() {
+        return (DefaultQueryHandler<Object, Object>) command ->
+                Mono.error(new RuntimeException("No Handler Registered"));
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    @SuppressWarnings("rawtypes")
+    public DefaultCommandHandler defaultCommandHandler() {
+        return message -> Mono.error(new RuntimeException("No Handler Registered"));
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public HandlerRegistry defaultHandlerRegistry() {
+        return HandlerRegistry.register();
+    }
+}
diff --git a/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/listeners/CommandsListenerConfig.java b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/listeners/CommandsListenerConfig.java
@@ -2,15 +2,16 @@
 
 
 import org.reactivecommons.async.commons.HandlerResolver;
-import org.reactivecommons.async.starter.config.ConnectionManager;
 import org.reactivecommons.async.starter.broker.BrokerProvider;
+import org.reactivecommons.async.starter.config.ConnectionManager;
 import org.reactivecommons.async.starter.config.DomainHandlers;
 import org.reactivecommons.async.starter.config.ReactiveCommonsConfig;
+import org.reactivecommons.async.starter.config.ReactiveCommonsListenersConfig;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 
 @Configuration
-@Import(ReactiveCommonsConfig.class)
+@Import({ReactiveCommonsConfig.class, ReactiveCommonsListenersConfig.class})
 public class CommandsListenerConfig extends AbstractListenerConfig {
 
     public CommandsListenerConfig(ConnectionManager manager, DomainHandlers handlers) {
diff --git a/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/listeners/EventsListenerConfig.java b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/listeners/EventsListenerConfig.java
@@ -6,11 +6,12 @@
 import org.reactivecommons.async.starter.broker.BrokerProvider;
 import org.reactivecommons.async.starter.config.DomainHandlers;
 import org.reactivecommons.async.starter.config.ReactiveCommonsConfig;
+import org.reactivecommons.async.starter.config.ReactiveCommonsListenersConfig;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 
 @Configuration
-@Import(ReactiveCommonsConfig.class)
+@Import({ReactiveCommonsConfig.class, ReactiveCommonsListenersConfig.class})
 public class EventsListenerConfig extends AbstractListenerConfig {
 
     public EventsListenerConfig(ConnectionManager manager, DomainHandlers handlers) {
diff --git a/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/listeners/NotificationEventsListenerConfig.java b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/listeners/NotificationEventsListenerConfig.java
@@ -6,11 +6,12 @@
 import org.reactivecommons.async.starter.broker.BrokerProvider;
 import org.reactivecommons.async.starter.config.DomainHandlers;
 import org.reactivecommons.async.starter.config.ReactiveCommonsConfig;
+import org.reactivecommons.async.starter.config.ReactiveCommonsListenersConfig;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 
 @Configuration
-@Import(ReactiveCommonsConfig.class)
+@Import({ReactiveCommonsConfig.class, ReactiveCommonsListenersConfig.class})
 public class NotificationEventsListenerConfig extends AbstractListenerConfig {
 
     public NotificationEventsListenerConfig(ConnectionManager manager, DomainHandlers handlers) {
diff --git a/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/listeners/QueriesListenerConfig.java b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/listeners/QueriesListenerConfig.java
@@ -2,15 +2,16 @@
 
 
 import org.reactivecommons.async.commons.HandlerResolver;
-import org.reactivecommons.async.starter.config.ConnectionManager;
 import org.reactivecommons.async.starter.broker.BrokerProvider;
+import org.reactivecommons.async.starter.config.ConnectionManager;
 import org.reactivecommons.async.starter.config.DomainHandlers;
 import org.reactivecommons.async.starter.config.ReactiveCommonsConfig;
+import org.reactivecommons.async.starter.config.ReactiveCommonsListenersConfig;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 
 @Configuration
-@Import(ReactiveCommonsConfig.class)
+@Import({ReactiveCommonsConfig.class, ReactiveCommonsListenersConfig.class})
 public class QueriesListenerConfig extends AbstractListenerConfig {
 
     public QueriesListenerConfig(ConnectionManager manager, DomainHandlers handlers) {
diff --git a/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/senders/DirectAsyncGatewayConfig.java b/starters/async-commons-starter/src/main/java/org/reactivecommons/async/starter/senders/DirectAsyncGatewayConfig.java
@@ -4,7 +4,6 @@
 import lombok.extern.java.Log;
 import org.reactivecommons.async.api.DirectAsyncGateway;
 import org.reactivecommons.async.starter.config.ConnectionManager;
-import org.reactivecommons.async.starter.config.DomainHandlers;
 import org.reactivecommons.async.starter.config.ReactiveCommonsConfig;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -20,10 +19,10 @@
 public class DirectAsyncGatewayConfig {
 
     @Bean
-    public DirectAsyncGateway genericDirectAsyncGateway(ConnectionManager manager, DomainHandlers handlers) {
+    public DirectAsyncGateway genericDirectAsyncGateway(ConnectionManager manager) {
         ConcurrentMap<String, DirectAsyncGateway> directAsyncGateways = new ConcurrentHashMap<>();
         manager.forDomain((domain, provider) -> directAsyncGateways.put(domain,
-                provider.getDirectAsyncGateway(handlers.get(domain))));
+                provider.getDirectAsyncGateway()));
         return new GenericDirectAsyncGateway(directAsyncGateways);
     }
 }
diff --git a/starters/async-commons-starter/src/test/java/org/reactivecommons/async/starter/mybroker/MyBrokerProvider.java b/starters/async-commons-starter/src/test/java/org/reactivecommons/async/starter/mybroker/MyBrokerProvider.java
@@ -52,7 +52,7 @@ public void listenQueries(HandlerResolver resolver) {
     }
 
     @Override
-    public void listenReplies(HandlerResolver resolver) {
+    public void listenReplies() {
         // for testing purposes
     }
 
diff --git a/starters/async-commons-starter/src/test/java/org/reactivecommons/async/starter/senders/DirectAsyncGatewayConfigTest.java b/starters/async-commons-starter/src/test/java/org/reactivecommons/async/starter/senders/DirectAsyncGatewayConfigTest.java
@@ -9,10 +9,8 @@
 import org.reactivecommons.async.commons.HandlerResolver;
 import org.reactivecommons.async.starter.broker.BrokerProvider;
 import org.reactivecommons.async.starter.config.ConnectionManager;
-import org.reactivecommons.async.starter.config.DomainHandlers;
 
 import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -40,14 +38,11 @@ void setUp() {
     @Test
     void shouldCreateAllDomainEventBuses() {
         // Arrange
-        when(brokerProvider.getDirectAsyncGateway(any())).thenReturn(domainEventBus);
-        DomainHandlers handlers = new DomainHandlers();
-        handlers.add("domain", resolver);
-        handlers.add("domain2", resolver);
+        when(brokerProvider.getDirectAsyncGateway()).thenReturn(domainEventBus);
         // Act
-        DirectAsyncGateway genericDomainEventBus = directAsyncGatewayConfig.genericDirectAsyncGateway(manager, handlers);
+        DirectAsyncGateway genericDomainEventBus = directAsyncGatewayConfig.genericDirectAsyncGateway(manager);
         // Assert
         assertNotNull(genericDomainEventBus);
-        verify(brokerProvider, times(2)).getDirectAsyncGateway(resolver);
+        verify(brokerProvider, times(2)).getDirectAsyncGateway();
     }
 }
diff --git a/starters/async-kafka-starter/src/main/java/org/reactivecommons/async/kafka/KafkaBrokerProvider.java b/starters/async-kafka-starter/src/main/java/org/reactivecommons/async/kafka/KafkaBrokerProvider.java
@@ -46,7 +46,7 @@ public DomainEventBus getDomainBus() {
     }
 
     @Override
-    public DirectAsyncGateway getDirectAsyncGateway(HandlerResolver resolver) {
+    public DirectAsyncGateway getDirectAsyncGateway() {
         return new KafkaDirectAsyncGateway();
     }
 
@@ -95,7 +95,7 @@ public void listenQueries(HandlerResolver resolver) {
     }
 
     @Override
-    public void listenReplies(HandlerResolver resolver) {
+    public void listenReplies() {
         // May be implemented in the future
     }
 
diff --git a/starters/async-kafka-starter/src/test/java/org/reactivecommons/async/kafka/KafkaBrokerProviderTest.java b/starters/async-kafka-starter/src/test/java/org/reactivecommons/async/kafka/KafkaBrokerProviderTest.java
@@ -99,7 +99,7 @@ void shouldCreateDomainEventBus() {
     @Test
     void shouldCreateDirectAsyncGateway() {
         // Act
-        DirectAsyncGateway domainBus = brokerProvider.getDirectAsyncGateway(handlerResolver);
+        DirectAsyncGateway domainBus = brokerProvider.getDirectAsyncGateway();
         // Assert
         assertThat(domainBus).isExactlyInstanceOf(KafkaDirectAsyncGateway.class);
     }
diff --git a/starters/async-rabbit-starter/src/main/java/org/reactivecommons/async/rabbit/RabbitMQBrokerProvider.java b/starters/async-rabbit-starter/src/main/java/org/reactivecommons/async/rabbit/RabbitMQBrokerProvider.java
@@ -53,12 +53,12 @@ public DomainEventBus getDomainBus() {
     }
 
     @Override
-    public DirectAsyncGateway getDirectAsyncGateway(HandlerResolver resolver) {
+    public DirectAsyncGateway getDirectAsyncGateway() {
         String exchangeName = props.getBrokerConfigProps().getDirectMessagesExchangeName();
         if (props.getCreateTopology()) {
             sender.getTopologyCreator().declare(exchange(exchangeName).durable(true).type("direct")).subscribe();
         }
-        listenReplies(resolver);
+        listenReplies();
         return new RabbitDirectAsyncGateway(config, router, sender, exchangeName, converter, meterRegistry);
     }
 
@@ -141,7 +141,7 @@ public void listenQueries(HandlerResolver resolver) {
     }
 
     @Override
-    public void listenReplies(HandlerResolver resolver) {
+    public void listenReplies() {
         if (props.isListenReplies()) {
             final ApplicationReplyListener replyListener =
                     new ApplicationReplyListener(router,
diff --git a/starters/async-rabbit-starter/src/main/java/org/reactivecommons/async/rabbit/RabbitMQSetupUtils.java b/starters/async-rabbit-starter/src/main/java/org/reactivecommons/async/rabbit/RabbitMQSetupUtils.java
@@ -150,8 +150,8 @@ private static Mono<Connection> createConnectionMono(ConnectionFactory factory,
                                                          String connectionType) {
         return Mono.fromCallable(() -> factory.newConnection(connectionPrefix + " " + connectionType))
                 .doOnError(err ->
-                        log.log(Level.SEVERE, "Error creating connection to RabbitMQ Broker in host" +
-                                factory.getHost() + ". Starting retry process...", err)
+                        log.log(Level.SEVERE, "Error creating connection to RabbitMQ Broker in host '" +
+                                factory.getHost() + "'. Starting retry process...", err)
                 )
                 .retryWhen(Retry.backoff(Long.MAX_VALUE, Duration.ofMillis(START_INTERVAL))
                         .maxBackoff(Duration.ofMillis(MAX_BACKOFF_INTERVAL)))
diff --git a/starters/async-rabbit-starter/src/test/java/org/reactivecommons/async/rabbit/RabbitMQBrokerProviderTest.java b/starters/async-rabbit-starter/src/test/java/org/reactivecommons/async/rabbit/RabbitMQBrokerProviderTest.java
@@ -112,7 +112,7 @@ void shouldCreateDirectAsyncGateway() {
         when(listener.getReceiver()).thenReturn(receiver);
         when(receiver.consumeAutoAck(any(String.class))).thenReturn(Flux.never());
         // Act
-        DirectAsyncGateway domainBus = brokerProvider.getDirectAsyncGateway(handlerResolver);
+        DirectAsyncGateway domainBus = brokerProvider.getDirectAsyncGateway();
         // Assert
         assertThat(domainBus).isExactlyInstanceOf(RabbitDirectAsyncGateway.class);
     }

Original file line number	Diff line number	Diff line change
`@@ -52,7 +52,7 @@ public void listenQueries(HandlerResolver resolver) {`
`52`	`52`	`}`
`53`	`53`
`54`	`54`	`@Override`
`55`		`- public void listenReplies(HandlerResolver resolver) {`
	`55`	`+ public void listenReplies() {`
`56`	`56`	`// for testing purposes`
`57`	`57`	`}`
`58`	`58`