Merge pull request #91 from reactive-commons/fix/circular-beans-dependency

juancgalvis · web-flow · commit ca48796ad30f · 2023-10-17T12:28:41.000-05:00
fix circular dependency
diff --git a/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/CommandListenersConfig.java b/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/CommandListenersConfig.java
@@ -24,9 +24,10 @@ public class CommandListenersConfig {
 
     @Bean
     public ApplicationCommandListener applicationCommandListener(ConnectionManager manager,
+                                                                 DomainHandlers handlers,
                                                                  MessageConverter converter,
                                                                  CustomReporter errorReporter) {
-        ApplicationCommandListener commandListener = new ApplicationCommandListener(manager.getListener(DEFAULT_DOMAIN), appName, manager.getHandlerResolver(DEFAULT_DOMAIN),
+        ApplicationCommandListener commandListener = new ApplicationCommandListener(manager.getListener(DEFAULT_DOMAIN), appName, handlers.get(DEFAULT_DOMAIN),
                 asyncProps.getDirect().getExchange(), converter, asyncProps.getWithDLQRetry(), asyncProps.getMaxRetries(),
                 asyncProps.getRetryDelay(), asyncProps.getDirect().getMaxLengthBytes(), manager.getDiscardNotifier(DEFAULT_DOMAIN), errorReporter);
 
diff --git a/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/ConnectionManager.java b/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/ConnectionManager.java
@@ -4,7 +4,6 @@
 import lombok.Getter;
 import lombok.Setter;
 import org.reactivecommons.async.commons.DiscardNotifier;
-import org.reactivecommons.async.rabbit.HandlerResolver;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageListener;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageSender;
 
@@ -20,7 +19,6 @@ public class ConnectionManager {
     public static class DomainConnections {
         private final ReactiveMessageListener listener;
         private final ReactiveMessageSender sender;
-        private final HandlerResolver handlerResolver;
         private final ConnectionFactoryProvider provider;
         @Setter
         private DiscardNotifier discardNotifier;
@@ -35,35 +33,39 @@ public void forListener(BiConsumer<String, ReactiveMessageListener> consumer) {
     }
 
     public void setDiscardNotifier(String domain, DiscardNotifier discardNotifier) {
-        connections.get(domain).setDiscardNotifier(discardNotifier);
+        getChecked(domain).setDiscardNotifier(discardNotifier);
     }
 
-    public ConnectionManager addDomain(String domain, ReactiveMessageListener listener, ReactiveMessageSender sender,
-                                       HandlerResolver resolver) {
+    public ConnectionManager addDomain(String domain, ReactiveMessageListener listener, ReactiveMessageSender sender) {
         connections.put(domain, DomainConnections.builder()
                 .listener(listener)
                 .sender(sender)
-                .handlerResolver(resolver)
                 .build());
         return this;
     }
 
     public ReactiveMessageSender getSender(String domain) {
-        return connections.get(domain).getSender();
+        return getChecked(domain).getSender();
     }
 
     public ReactiveMessageListener getListener(String domain) {
-        return connections.get(domain).getListener();
+        return getChecked(domain).getListener();
     }
 
-    public DiscardNotifier getDiscardNotifier(String domain) {
-        return connections.get(domain).getDiscardNotifier();
+    private DomainConnections getChecked(String domain) {
+        DomainConnections domainConnections = connections.get(domain);
+        if (domainConnections == null) {
+            throw new RuntimeException("You are trying to use the domain " + domain
+                    + " but this connection is not defined");
+        }
+        return domainConnections;
     }
 
-    public HandlerResolver getHandlerResolver(String domain) {
-        return connections.get(domain).getHandlerResolver();
+    public DiscardNotifier getDiscardNotifier(String domain) {
+        return getChecked(domain).getDiscardNotifier();
     }
+
     public ConnectionFactoryProvider getProvider(String domain) {
-        return connections.get(domain).getProvider();
+        return getChecked(domain).getProvider();
     }
 }
diff --git a/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/DomainHandlers.java b/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/DomainHandlers.java
@@ -0,0 +1,23 @@
+package org.reactivecommons.async.rabbit.config;
+
+import org.reactivecommons.async.rabbit.HandlerResolver;
+
+import java.util.Map;
+import java.util.TreeMap;
+
+public class DomainHandlers {
+    private final Map<String, HandlerResolver> handlers = new TreeMap<>();
+
+    public void add(String domain, HandlerResolver resolver) {
+        this.handlers.put(domain, resolver);
+    }
+
+    public HandlerResolver get(String domain) {
+        HandlerResolver handlerResolver = handlers.get(domain);
+        if (handlerResolver == null) {
+            throw new RuntimeException("You are trying to use the domain " + domain
+                    + " but this connection is not defined");
+        }
+        return handlerResolver;
+    }
+}
diff --git a/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/EventListenersConfig.java b/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/EventListenersConfig.java
@@ -26,12 +26,13 @@ public class EventListenersConfig {
 
     @Bean
     public ApplicationEventListener eventListener(MessageConverter messageConverter,
-                                                  ConnectionManager manager, CustomReporter errorReporter) {
+                                                  ConnectionManager manager, DomainHandlers handlers,
+                                                  CustomReporter errorReporter) {
         AtomicReference<ApplicationEventListener> external = new AtomicReference<>();
         manager.forListener((domain, receiver) -> {
             final ApplicationEventListener listener = new ApplicationEventListener(receiver,
                     appName + ".subsEvents",
-                    manager.getHandlerResolver(domain),
+                    handlers.get(domain),
                     asyncProps.getDomain().getEvents().getExchange(),
                     messageConverter, asyncProps.getWithDLQRetry(),
                     asyncProps.getMaxRetries(),
diff --git a/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/HandlerResolverBuilder.java b/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/HandlerResolverBuilder.java
@@ -2,6 +2,7 @@
 
 import lombok.AccessLevel;
 import lombok.NoArgsConstructor;
+import lombok.extern.log4j.Log4j2;
 import org.reactivecommons.async.api.DefaultCommandHandler;
 import org.reactivecommons.async.api.HandlerRegistry;
 import org.reactivecommons.async.api.handlers.registered.RegisteredCommandHandler;
@@ -17,6 +18,7 @@
 import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_DOMAIN;
 
 @NoArgsConstructor(access = AccessLevel.PRIVATE)
+@Log4j2
 public class HandlerResolverBuilder {
 
     public static HandlerResolver buildResolver(String domain,
@@ -75,7 +77,13 @@ private static ConcurrentMap<String, RegisteredEventListener<?>> getEventHandler
         // event handlers and dynamic handlers
         return registries
                 .values().stream()
-                .flatMap(r -> Stream.concat(r.getDomainEventListeners().get(domain).stream(), getDynamics(domain, r)))
+                .flatMap(r -> {
+                    if (r.getDomainEventListeners().containsKey(domain)) {
+                        return Stream.concat(r.getDomainEventListeners().get(domain).stream(), getDynamics(domain, r));
+                    }
+                    log.warn("Domain " + domain + "does not have a connection defined in your configuration and you want to listen from it");
+                    return Stream.empty();
+                })
                 .collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler),
                         ConcurrentHashMap::putAll);
     }
@@ -90,7 +98,13 @@ private static Stream<RegisteredEventListener<?>> getDynamics(String domain, Han
     private static ConcurrentMap<String, RegisteredEventListener<?>> getEventsToBind(String domain, Map<String, HandlerRegistry> registries) {
         return registries
                 .values().stream()
-                .flatMap(r -> r.getDomainEventListeners().get(domain).stream())
+                .flatMap(r -> {
+                    if (r.getDomainEventListeners().containsKey(domain)) {
+                        return r.getDomainEventListeners().get(domain).stream();
+                    }
+                    log.warn("Domain " + domain + "does not have a connection defined in your configuration and you want to listen from it");
+                    return Stream.empty();
+                })
                 .collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler),
                         ConcurrentHashMap::putAll);
     }
diff --git a/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/NotificacionListenersConfig.java b/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/NotificacionListenersConfig.java
@@ -24,13 +24,14 @@ public class NotificacionListenersConfig {
 
     @Bean
     public ApplicationNotificationListener eventNotificationListener(ConnectionManager manager,
+                                                                     DomainHandlers handlers,
                                                                      MessageConverter messageConverter,
                                                                      CustomReporter errorReporter) {
         final ApplicationNotificationListener listener = new ApplicationNotificationListener(
                 manager.getListener(DEFAULT_DOMAIN),
                 asyncProps.getDomain().getEvents().getExchange(),
                 asyncProps.getNotificationProps().getQueueName(appName),
-                manager.getHandlerResolver(DEFAULT_DOMAIN),
+                handlers.get(DEFAULT_DOMAIN),
                 messageConverter,
                 manager.getDiscardNotifier(DEFAULT_DOMAIN),
                 errorReporter);
diff --git a/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/QueryListenerConfig.java b/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/QueryListenerConfig.java
@@ -24,10 +24,11 @@ public class QueryListenerConfig {
 
     @Bean
     public ApplicationQueryListener queryListener(MessageConverter converter,
+                                                  DomainHandlers handlers,
                                                   ConnectionManager manager,
                                                   CustomReporter errorReporter) {
         final ApplicationQueryListener listener = new ApplicationQueryListener(manager.getListener(DEFAULT_DOMAIN),
-                appName + ".query", manager.getHandlerResolver(DEFAULT_DOMAIN), manager.getSender(DEFAULT_DOMAIN), asyncProps.getDirect().getExchange(), converter,
+                appName + ".query", handlers.get(DEFAULT_DOMAIN), manager.getSender(DEFAULT_DOMAIN), asyncProps.getDirect().getExchange(), converter,
                 asyncProps.getGlobal().getExchange(), asyncProps.getWithDLQRetry(), asyncProps.getMaxRetries(),
                 asyncProps.getRetryDelay(), asyncProps.getGlobal().getMaxLengthBytes(),
                 asyncProps.getDirect().isDiscardTimeoutQueries(), manager.getDiscardNotifier(DEFAULT_DOMAIN), errorReporter);
diff --git a/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/RabbitMqConfig.java b/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/RabbitMqConfig.java
@@ -60,25 +60,34 @@ public class RabbitMqConfig {
     public ConnectionManager buildConnectionManager(@Value("${spring.application.name}") String appName,
                                                     AsyncProps props,
                                                     RabbitProperties defaultAppProps,
-                                                    MessageConverter converter,
-                                                    ApplicationContext context,
-                                                    HandlerRegistry primaryRegistry,
-                                                    DefaultCommandHandler<?> commandHandler) {
+                                                    MessageConverter converter) {
         ConnectionManager connectionManager = new ConnectionManager();
+        props.getConnections().computeIfAbsent(DEFAULT_DOMAIN, k -> defaultAppProps);
+        props.getConnections()
+                .forEach((domain, properties) -> {
+                    ConnectionFactoryProvider provider = createConnectionFactoryProvider(properties);
+                    ReactiveMessageSender sender = createMessageSender(appName, provider, properties, converter);
+                    ReactiveMessageListener listener = createMessageListener(appName, provider, props);
+                    connectionManager.addDomain(domain, listener, sender);
+                });
+        return connectionManager;
+    }
+
+    @Bean
+    public DomainHandlers buildHandlers(AsyncProps props, RabbitProperties defaultAppProps, 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);
         }
         props.getConnections().computeIfAbsent(DEFAULT_DOMAIN, k -> defaultAppProps);
         props.getConnections()
                 .forEach((domain, properties) -> {
-                    ConnectionFactoryProvider provider = createConnectionFactoryProvider(properties);
-                    ReactiveMessageSender sender = createMessageSender(appName, provider, properties, converter);
-                    ReactiveMessageListener listener = createMessageListener(appName, provider, props);
                     HandlerResolver resolver = HandlerResolverBuilder.buildResolver(domain, registries, commandHandler);
-                    connectionManager.addDomain(domain, listener, sender, resolver);
+                    handlers.add(domain, resolver);
                 });
-        return connectionManager;
+        return handlers;
     }
 
 
@@ -187,8 +196,9 @@ Mono<Connection> createConnectionMono(ConnectionFactory factory, String connecti
     }
 
     @Bean
-    public DynamicRegistry dynamicRegistry(ConnectionManager connectionManager, IBrokerConfigProps props) {
-        return new DynamicRegistryImp(connectionManager.getHandlerResolver(DEFAULT_DOMAIN),
+    public DynamicRegistry dynamicRegistry(ConnectionManager connectionManager, DomainHandlers handlers,
+                                           IBrokerConfigProps props) {
+        return new DynamicRegistryImp(handlers.get(DEFAULT_DOMAIN),
                 connectionManager.getListener(DEFAULT_DOMAIN).getTopologyCreator(), props);
     }
 
diff --git a/async/async-rabbit-starter-eda/src/test/java/org/reactivecommons/async/rabbit/config/CommandListenersConfigTest.java b/async/async-rabbit-starter-eda/src/test/java/org/reactivecommons/async/rabbit/config/CommandListenersConfigTest.java
@@ -40,6 +40,7 @@ class CommandListenersConfigTest {
     private final CustomReporter customReporter = mock(CustomReporter.class);
     private final Receiver receiver = mock(Receiver.class);
     private final ConnectionManager manager = new ConnectionManager();
+    private final DomainHandlers handlers = new DomainHandlers();
 
     @BeforeEach
     public void init() throws NoSuchFieldException, IllegalAccessException {
@@ -53,12 +54,13 @@ public void init() throws NoSuchFieldException, IllegalAccessException {
         when(receiver.consumeManualAck(any(String.class), any(ConsumeOptions.class))).thenReturn(Flux.never());
         when(listener.getReceiver()).thenReturn(receiver);
         when(listener.getMaxConcurrency()).thenReturn(20);
-        manager.addDomain(DEFAULT_DOMAIN, listener, null, handlerResolver);
+        manager.addDomain(DEFAULT_DOMAIN, listener, null);
+        handlers.add(DEFAULT_DOMAIN, handlerResolver);
     }
 
     @Test
     void applicationCommandListener() {
-        final ApplicationCommandListener commandListener = config.applicationCommandListener(manager, messageConverter, customReporter);
+        final ApplicationCommandListener commandListener = config.applicationCommandListener(manager, handlers, messageConverter, customReporter);
         Assertions.assertThat(commandListener).isNotNull();
     }
 }
diff --git a/async/async-rabbit-starter-eda/src/test/java/org/reactivecommons/async/rabbit/config/EventListenersConfigTest.java b/async/async-rabbit-starter-eda/src/test/java/org/reactivecommons/async/rabbit/config/EventListenersConfigTest.java
@@ -39,6 +39,7 @@ class EventListenersConfigTest {
     private final CustomReporter customReporter = mock(CustomReporter.class);
     private final Receiver receiver = mock(Receiver.class);
     private ConnectionManager connectionManager;
+    private final DomainHandlers handlers = new DomainHandlers();
 
     @BeforeEach
     public void init() {
@@ -54,14 +55,16 @@ public void init() {
         when(listener.getReceiver()).thenReturn(receiver);
         when(listener.getMaxConcurrency()).thenReturn(20);
         connectionManager = new ConnectionManager();
-        connectionManager.addDomain(HandlerRegistry.DEFAULT_DOMAIN, listener, sender, handlerResolver);
+        connectionManager.addDomain(HandlerRegistry.DEFAULT_DOMAIN, listener, sender);
+        handlers.add(HandlerRegistry.DEFAULT_DOMAIN, handlerResolver);
     }
 
     @Test
     void eventListener() {
         final ApplicationEventListener eventListener = config.eventListener(
                 messageConverter,
                 connectionManager,
+                handlers,
                 customReporter
         );
 
diff --git a/async/async-rabbit-starter-eda/src/test/java/org/reactivecommons/async/rabbit/config/NotificacionListenersConfigTest.java b/async/async-rabbit-starter-eda/src/test/java/org/reactivecommons/async/rabbit/config/NotificacionListenersConfigTest.java
@@ -39,6 +39,7 @@ class NotificacionListenersConfigTest {
     private final CustomReporter customReporter = mock(CustomReporter.class);
     private final Receiver receiver = mock(Receiver.class);
     private final ConnectionManager manager = new ConnectionManager();
+    private final DomainHandlers handlers = new DomainHandlers();
 
     @BeforeEach
     public void init() {
@@ -53,13 +54,14 @@ public void init() {
         when(receiver.consumeManualAck(any(String.class), any(ConsumeOptions.class))).thenReturn(Flux.never());
         when(listener.getReceiver()).thenReturn(receiver);
         when(listener.getMaxConcurrency()).thenReturn(20);
-        manager.addDomain(DEFAULT_DOMAIN, listener, null, handlerResolver);
+        manager.addDomain(DEFAULT_DOMAIN, listener, null);
+        handlers.add(DEFAULT_DOMAIN, handlerResolver);
     }
 
     @Test
     void eventNotificationListener() {
         final ApplicationNotificationListener applicationEventListener =
-                config.eventNotificationListener(manager, messageConverter, customReporter);
+                config.eventNotificationListener(manager, handlers, messageConverter, customReporter);
         Assertions.assertThat(applicationEventListener).isNotNull();
     }
 }
diff --git a/async/async-rabbit-starter-eda/src/test/java/org/reactivecommons/async/rabbit/config/QueryListenerConfigTest.java b/async/async-rabbit-starter-eda/src/test/java/org/reactivecommons/async/rabbit/config/QueryListenerConfigTest.java
@@ -39,6 +39,7 @@ class QueryListenerConfigTest {
     private final Receiver receiver = mock(Receiver.class);
     private final ReactiveMessageSender sender = mock(ReactiveMessageSender.class);
     private final ConnectionManager manager = new ConnectionManager();
+    private final DomainHandlers handlers = new DomainHandlers();
 
     @BeforeEach
     public void init() {
@@ -53,12 +54,13 @@ public void init() {
         when(receiver.consumeManualAck(any(String.class), any(ConsumeOptions.class))).thenReturn(Flux.never());
         when(listener.getReceiver()).thenReturn(receiver);
         when(listener.getMaxConcurrency()).thenReturn(20);
-        manager.addDomain(DEFAULT_DOMAIN, listener, sender, handlerResolver);
+        manager.addDomain(DEFAULT_DOMAIN, listener, sender);
+        handlers.add(DEFAULT_DOMAIN, handlerResolver);
     }
 
     @Test
     void queryListener() {
-        final ApplicationQueryListener queryListener = config.queryListener(messageConverter, manager, customReporter);
+        final ApplicationQueryListener queryListener = config.queryListener(messageConverter, handlers, manager, customReporter);
         Assertions.assertThat(queryListener).isNotNull();
     }
 }
diff --git a/gradle.properties b/gradle.properties
@@ -1,4 +1,4 @@
-version=2.0.2
+version=2.0.3
 springBootVersion=3.0.2
 reactorRabbitVersion=1.5.5
 gradleVersionsVersion=0.36.0
diff --git a/samples/async/receiver-responder/src/main/java/sample/SampleReceiverApp.java b/samples/async/receiver-responder/src/main/java/sample/SampleReceiverApp.java
diff --git a/samples/async/receiver-responder/src/main/java/sample/UseCase.java b/samples/async/receiver-responder/src/main/java/sample/UseCase.java
diff --git a/samples/async/receiver-responder/src/main/resources/application.yaml b/samples/async/receiver-responder/src/main/resources/application.yaml
diff --git a/samples/async/sender-client/src/main/resources/application.yaml b/samples/async/sender-client/src/main/resources/application.yaml
