update health for eda, append default app props

juancgalvis · juancgalvis · commit 151855100207 · 2023-06-27T10:55:03.000-05:00
diff --git a/async/async-commons-api/src/main/java/org/reactivecommons/async/api/DirectAsyncGateway.java b/async/async-commons-api/src/main/java/org/reactivecommons/async/api/DirectAsyncGateway.java
@@ -6,8 +6,20 @@
 
 public interface DirectAsyncGateway {
     <T> Mono<Void> sendCommand(Command<T> command, String targetName);
+
+    <T> Mono<Void> sendCommand(Command<T> command, String targetName, String domain);
+
     Mono<Void> sendCommand(CloudEvent command, String targetName);
+
+    Mono<Void> sendCommand(CloudEvent command, String targetName, String domain);
+
     <T, R> Mono<R> requestReply(AsyncQuery<T> query, String targetName, Class<R> type);
+
+    <T, R> Mono<R> requestReply(AsyncQuery<T> query, String targetName, Class<R> type, String domain);
+
     <R extends CloudEvent> Mono<R> requestReply(CloudEvent query, String targetName, Class<R> type);
+
+    <R extends CloudEvent> Mono<R> requestReply(CloudEvent query, String targetName, Class<R> type, String domain);
+
     <T> Mono<Void> reply(T response, From from);
 }
diff --git a/async/async-commons-api/src/main/java/org/reactivecommons/async/api/HandlerRegistry.java b/async/async-commons-api/src/main/java/org/reactivecommons/async/api/HandlerRegistry.java
@@ -23,7 +23,7 @@
 @Getter
 @NoArgsConstructor(access = AccessLevel.PACKAGE)
 public class HandlerRegistry {
-    public static final String DEFAULT_LISTENER = "app";
+    public static final String DEFAULT_DOMAIN = "app";
     private final Map<String, List<RegisteredEventListener<?>>> domainEventListeners = new ConcurrentHashMap<>();
     private final List<RegisteredEventListener<?>> dynamicEventHandlers = new CopyOnWriteArrayList<>();
     private final List<RegisteredEventListener<?>> eventNotificationListener = new CopyOnWriteArrayList<>();
@@ -33,7 +33,7 @@ public class HandlerRegistry {
 
     public static HandlerRegistry register() {
         HandlerRegistry instance = new HandlerRegistry();
-        instance.domainEventListeners.put(DEFAULT_LISTENER, new CopyOnWriteArrayList<>());
+        instance.domainEventListeners.put(DEFAULT_DOMAIN, new CopyOnWriteArrayList<>());
         return instance;
     }
 
@@ -44,7 +44,7 @@ public <T> HandlerRegistry listenDomainEvent(String domain, String eventName, Ev
     }
 
     public <T> HandlerRegistry listenEvent(String eventName, EventHandler<T> handler, Class<T> eventClass) {
-        domainEventListeners.computeIfAbsent(DEFAULT_LISTENER, ignored -> new CopyOnWriteArrayList<>())
+        domainEventListeners.computeIfAbsent(DEFAULT_DOMAIN, ignored -> new CopyOnWriteArrayList<>())
         .add(new RegisteredEventListener<>(eventName, handler, eventClass));
         return this;
     }
diff --git a/async/async-commons-api/src/test/java/org/reactivecommons/async/api/HandlerRegistryTest.java b/async/async-commons-api/src/test/java/org/reactivecommons/async/api/HandlerRegistryTest.java
@@ -16,7 +16,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
-import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_LISTENER;
+import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_DOMAIN;
 
 class HandlerRegistryTest {
     private final HandlerRegistry registry = HandlerRegistry.register();
@@ -28,7 +28,7 @@ void shouldListenEventWithTypeInferenceWhenClassInstanceIsUsed() {
 
         registry.listenEvent(name, eventHandler);
 
-        assertThat(registry.getDomainEventListeners().get(DEFAULT_LISTENER))
+        assertThat(registry.getDomainEventListeners().get(DEFAULT_DOMAIN))
                 .anySatisfy(registered -> assertThat(registered)
                         .extracting(RegisteredEventListener::getPath, RegisteredEventListener::getInputClass, RegisteredEventListener::getHandler)
                         .containsExactly(name, SomeDataClass.class, eventHandler)).hasSize(1);
@@ -44,7 +44,7 @@ void shouldRegisterPatternEventHandlerWithTypeInference() {
         RegisteredEventListener<SomeDataClass> expectedRegisteredEventListener =
                 new RegisteredEventListener<>(eventNamePattern, eventHandler, SomeDataClass.class);
 
-        assertThat(registry.getDomainEventListeners().get(DEFAULT_LISTENER))
+        assertThat(registry.getDomainEventListeners().get(DEFAULT_DOMAIN))
                 .anySatisfy(registeredEventListener -> assertThat(registeredEventListener)
                         .usingRecursiveComparison()
                         .isEqualTo(expectedRegisteredEventListener));
@@ -63,7 +63,7 @@ void shouldRegisterPatternEventHandler() {
         RegisteredEventListener<SomeDataClass> expectedRegisteredEventListener =
                 new RegisteredEventListener<>(eventNamePattern, eventHandler, SomeDataClass.class);
 
-        assertThat(registry.getDomainEventListeners().get(DEFAULT_LISTENER))
+        assertThat(registry.getDomainEventListeners().get(DEFAULT_DOMAIN))
                 .anySatisfy(registeredEventListener -> assertThat(registeredEventListener)
                         .usingRecursiveComparison()
                         .isEqualTo(expectedRegisteredEventListener));
@@ -85,7 +85,7 @@ public void listenEvent() {
         EventHandler<SomeDataClass> handler = mock(EventHandler.class);
         registry.listenEvent(name, handler, SomeDataClass.class);
 
-        assertThat(registry.getDomainEventListeners().get(DEFAULT_LISTENER))
+        assertThat(registry.getDomainEventListeners().get(DEFAULT_DOMAIN))
                 .anySatisfy(registered -> assertThat(registered)
                         .extracting(RegisteredEventListener::getPath, RegisteredEventListener::getInputClass, RegisteredEventListener::getHandler)
                         .containsExactly(name, SomeDataClass.class, handler)).hasSize(1);
diff --git a/async/async-commons/src/test/java/org/reactivecommons/async/commons/utils/matcher/KeyMatcherPerformanceWildcardTest.java b/async/async-commons/src/test/java/org/reactivecommons/async/commons/utils/matcher/KeyMatcherPerformanceWildcardTest.java
@@ -8,7 +8,12 @@
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Paths;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 class KeyMatcherPerformanceWildcardTest {
@@ -27,13 +32,13 @@ public void setUp() {
         File file = new File(classLoader.getResource("wildcard_names_for_matching.txt").getFile());
         File file2 = new File(classLoader.getResource("concrete_names_for_matching.txt").getFile());
         try {
-             Set<String> names =  new HashSet<>(Files
+            Set<String> names = new HashSet<>(Files
                     .readAllLines(Paths.get(file.getAbsolutePath())));
             candidates = names.stream()
                     .collect(Collectors.toMap(name -> name, name -> name));
             testList = new ArrayList<>(new HashSet<>(Files
-                .readAllLines(Paths.get(file2.getAbsolutePath()))));
-            testResultList = new ArrayList<>(testList.size()*10);
+                    .readAllLines(Paths.get(file2.getAbsolutePath()))));
+            testResultList = new ArrayList<>(testList.size() * 10);
         } catch (IOException e) {
             e.printStackTrace();
         }
@@ -43,14 +48,14 @@ public void setUp() {
     void keyMatcherLookupShouldPerformInLessThan30Micros() {
         final int size = testList.size();
         final long init = System.currentTimeMillis();
-        for (int i = 0; i< size*10; ++i){
-            testResultList.add(keyMatcher.match(candidates.keySet(), testList.get(i%size)));
+        for (int i = 0; i < size * 10; ++i) {
+            testResultList.add(keyMatcher.match(candidates.keySet(), testList.get(i % size)));
         }
         final long end = System.currentTimeMillis();
 
 
         final long total = end - init;
-        final double microsPerLookup =  ((total+0.0)/testResultList.size())*1000;
+        final double microsPerLookup = ((total + 0.0) / testResultList.size()) * 1000;
         System.out.println("Performed Lookups: " + testResultList.size());
         System.out.println("Total Execution Time: " + total + "ms");
         System.out.println("Microseconds per lookup: " + microsPerLookup + "us");
diff --git a/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/RabbitEDADirectAsyncGateway.java b/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/RabbitEDADirectAsyncGateway.java
@@ -0,0 +1,25 @@
+package org.reactivecommons.async;
+
+import io.micrometer.core.instrument.MeterRegistry;
+import org.reactivecommons.async.commons.config.BrokerConfig;
+import org.reactivecommons.async.commons.converters.MessageConverter;
+import org.reactivecommons.async.commons.reply.ReactiveReplyRouter;
+import org.reactivecommons.async.rabbit.RabbitDirectAsyncGateway;
+import org.reactivecommons.async.rabbit.communications.ReactiveMessageSender;
+import org.reactivecommons.async.rabbit.config.ConnectionManager;
+
+import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_DOMAIN;
+
+public class RabbitEDADirectAsyncGateway extends RabbitDirectAsyncGateway {
+    private final ConnectionManager manager;
+
+    public RabbitEDADirectAsyncGateway(BrokerConfig config, ReactiveReplyRouter router, ConnectionManager manager, String exchange, MessageConverter converter, MeterRegistry meterRegistry) {
+        super(config, router, manager.getSender(DEFAULT_DOMAIN), exchange, converter, meterRegistry);
+        this.manager = manager;
+    }
+
+    @Override
+    protected ReactiveMessageSender resolveSender(String domain) {
+        return manager.getSender(domain);
+    }
+}
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
@@ -21,6 +21,7 @@ public static class DomainConnections {
         private final ReactiveMessageListener listener;
         private final ReactiveMessageSender sender;
         private final HandlerResolver handlerResolver;
+        private final ConnectionFactoryProvider provider;
         @Setter
         private DiscardNotifier discardNotifier;
     }
@@ -62,4 +63,7 @@ public DiscardNotifier getDiscardNotifier(String domain) {
     public HandlerResolver getHandlerResolver(String domain) {
         return connections.get(domain).getHandlerResolver();
     }
+    public ConnectionFactoryProvider getProvider(String domain) {
+        return connections.get(domain).getProvider();
+    }
 }
diff --git a/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/DirectAsyncGatewayConfig.java b/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/DirectAsyncGatewayConfig.java
@@ -7,16 +7,14 @@
 import org.reactivecommons.async.commons.converters.MessageConverter;
 import org.reactivecommons.async.commons.reply.ReactiveReplyRouter;
 import org.reactivecommons.async.rabbit.RabbitDirectAsyncGateway;
-import org.reactivecommons.async.rabbit.communications.ReactiveMessageListener;
-import org.reactivecommons.async.rabbit.communications.ReactiveMessageSender;
 import org.reactivecommons.async.rabbit.config.props.BrokerConfigProps;
 import org.reactivecommons.async.rabbit.listeners.ApplicationReplyListener;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 
-import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_LISTENER;
+import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_DOMAIN;
 
 @Configuration
 @Import(RabbitMqConfig.class)
@@ -27,12 +25,12 @@ public class DirectAsyncGatewayConfig {
 
     @Bean
     public RabbitDirectAsyncGateway rabbitDirectAsyncGateway(BrokerConfig config, ReactiveReplyRouter router, ConnectionManager manager, MessageConverter converter, MeterRegistry meterRegistry) throws Exception {
-        return new RabbitDirectAsyncGateway(config, router, manager.getSender(DEFAULT_LISTENER), props.getDirectMessagesExchangeName(), converter, meterRegistry);
+        return new RabbitDirectAsyncGateway(config, router, manager.getSender(DEFAULT_DOMAIN), props.getDirectMessagesExchangeName(), converter, meterRegistry);
     }
 
     @Bean
     public ApplicationReplyListener msgListener(ReactiveReplyRouter router, BrokerConfig config, ConnectionManager manager) {
-        final ApplicationReplyListener replyListener = new ApplicationReplyListener(router, manager.getListener(DEFAULT_LISTENER), props.getReplyQueue());
+        final ApplicationReplyListener replyListener = new ApplicationReplyListener(router, manager.getListener(DEFAULT_DOMAIN), props.getReplyQueue());
         replyListener.startListening(config.getRoutingKey());
         return replyListener;
     }
diff --git a/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/EventBusConfig.java b/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/EventBusConfig.java
@@ -12,7 +12,7 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 
-import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_LISTENER;
+import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_DOMAIN;
 import static reactor.rabbitmq.ExchangeSpecification.exchange;
 
 @Configuration
@@ -22,11 +22,11 @@ public class EventBusConfig {
     @Bean // app connection
     public DomainEventBus domainEventBus(ConnectionManager manager, BrokerConfigProps props, BrokerConfig config,
                                          ObjectMapperSupplier objectMapperSupplier) {
-        ReactiveMessageSender sender = manager.getSender(DEFAULT_LISTENER);
+        ReactiveMessageSender sender = manager.getSender(DEFAULT_DOMAIN);
         final String exchangeName = props.getDomainEventsExchangeName();
         sender.getTopologyCreator().declare(exchange(exchangeName).durable(true).type("topic")).subscribe();
         DomainEventBus domainEventBus = new RabbitDomainEventBus(sender, exchangeName, config);
-        manager.setDiscardNotifier(DEFAULT_LISTENER, createDiscardNotifier(domainEventBus, objectMapperSupplier));
+        manager.setDiscardNotifier(DEFAULT_DOMAIN, createDiscardNotifier(domainEventBus, objectMapperSupplier));
         return domainEventBus;
     }
 
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
@@ -12,7 +12,7 @@
 
 import java.util.concurrent.atomic.AtomicReference;
 
-import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_LISTENER;
+import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_DOMAIN;
 
 @Configuration
 @RequiredArgsConstructor
@@ -40,7 +40,7 @@ public ApplicationEventListener eventListener(MessageConverter messageConverter,
                     manager.getDiscardNotifier(domain),
                     errorReporter,
                     appName);
-            if (DEFAULT_LISTENER.equals(domain)) {
+            if (DEFAULT_DOMAIN.equals(domain)) {
                 external.set(listener);
             }
             listener.startListener();
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
@@ -14,7 +14,7 @@
 import java.util.concurrent.ConcurrentMap;
 import java.util.stream.Stream;
 
-import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_LISTENER;
+import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_DOMAIN;
 
 @NoArgsConstructor(access = AccessLevel.PRIVATE)
 public class HandlerResolverBuilder {
@@ -23,7 +23,7 @@ public static HandlerResolver buildResolver(String domain,
                                                 Map<String, HandlerRegistry> registries,
                                                 final DefaultCommandHandler defaultCommandHandler) {
 
-        if (DEFAULT_LISTENER.equals(domain)) {
+        if (DEFAULT_DOMAIN.equals(domain)) {
             final ConcurrentMap<String, RegisteredQueryHandler<?, ?>> queryHandlers = registries
                     .values().stream()
                     .flatMap(r -> r.getHandlers().stream())
@@ -81,7 +81,7 @@ private static ConcurrentMap<String, RegisteredEventListener<?>> getEventHandler
     }
 
     private static Stream<RegisteredEventListener<?>> getDynamics(String domain, HandlerRegistry r) {
-        if (DEFAULT_LISTENER.equals(domain)) {
+        if (DEFAULT_DOMAIN.equals(domain)) {
             return r.getDynamicEventHandlers().stream();
         }
         return Stream.of();
diff --git a/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/RabbitHealthConfig.java b/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/RabbitHealthConfig.java
@@ -6,12 +6,14 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
+import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_DOMAIN;
+
 @Configuration
 @ConditionalOnClass(AbstractReactiveHealthIndicator.class)
 public class RabbitHealthConfig {
 
     @Bean
-    public RabbitReactiveHealthIndicator rabbitHealthIndicator(ConnectionFactoryProvider provider) {
-        return new RabbitReactiveHealthIndicator(provider);
+    public RabbitReactiveHealthIndicator rabbitHealthIndicator(ConnectionManager manager) {
+        return new RabbitReactiveHealthIndicator(manager.getProvider(DEFAULT_DOMAIN)); // TODO: Check every domain connection
     }
 }
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
@@ -27,7 +27,6 @@
 import org.reactivecommons.async.rabbit.config.props.AsyncProps;
 import org.reactivecommons.async.rabbit.config.props.BrokerConfigProps;
 import org.reactivecommons.async.rabbit.converters.json.JacksonCloudEventMessageConverter;
-import org.reactivecommons.async.rabbit.converters.json.JacksonMessageConverter;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -52,7 +51,7 @@
 import java.util.Map;
 import java.util.logging.Level;
 
-import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_LISTENER;
+import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_DOMAIN;
 
 @Log
 @Configuration
@@ -72,11 +71,13 @@ public class RabbitMqConfig {
     @Bean
     public ConnectionManager buildConnectionManager(@Value("${spring.application.name}") String appName,
                                                     AsyncProps props,
+                                                    RabbitProperties defaultAppProps,
                                                     MessageConverter converter,
                                                     ApplicationContext context,
                                                     DefaultCommandHandler<?> commandHandler) {
         ConnectionManager connectionManager = new ConnectionManager();
         final Map<String, HandlerRegistry> registries = context.getBeansOfType(HandlerRegistry.class);
+        props.getConnections().computeIfAbsent(DEFAULT_DOMAIN, k -> defaultAppProps);
         props.getConnections()
                 .forEach((domain, properties) -> {
                     ConnectionFactoryProvider provider = createConnectionFactoryProvider(properties);
@@ -195,8 +196,8 @@ Mono<Connection> createConnectionMono(ConnectionFactory factory, String connecti
 
     @Bean
     public DynamicRegistry dynamicRegistry(ConnectionManager connectionManager, IBrokerConfigProps props) {
-        return new DynamicRegistryImp(connectionManager.getHandlerResolver(DEFAULT_LISTENER),
-                connectionManager.getListener(DEFAULT_LISTENER).getTopologyCreator(), props);
+        return new DynamicRegistryImp(connectionManager.getHandlerResolver(DEFAULT_DOMAIN),
+                connectionManager.getListener(DEFAULT_DOMAIN).getTopologyCreator(), props);
     }
 
     @Bean
diff --git a/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/props/AsyncProps.java b/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/props/AsyncProps.java
@@ -6,7 +6,6 @@
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.NestedConfigurationProperty;
 
-import java.util.HashMap;
 import java.util.Map;
 import java.util.TreeMap;
 
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
@@ -54,7 +54,7 @@ public void init() {
         when(listener.getReceiver()).thenReturn(receiver);
         when(listener.getMaxConcurrency()).thenReturn(20);
         connectionManager = new ConnectionManager();
-        connectionManager.addDomain(HandlerRegistry.DEFAULT_LISTENER, listener, sender, handlerResolver);
+        connectionManager.addDomain(HandlerRegistry.DEFAULT_DOMAIN, listener, sender, handlerResolver);
     }
 
     @Test
diff --git a/async/async-rabbit-starter/src/main/java/org/reactivecommons/async/rabbit/config/RabbitMqConfig.java b/async/async-rabbit-starter/src/main/java/org/reactivecommons/async/rabbit/config/RabbitMqConfig.java
@@ -62,7 +62,7 @@
 import java.util.logging.Level;
 import java.util.stream.Stream;
 
-import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_LISTENER;
+import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_DOMAIN;
 import static reactor.rabbitmq.ExchangeSpecification.exchange;
 
 @Log
@@ -214,14 +214,14 @@ public HandlerResolver resolver(ApplicationContext context, DefaultCommandHandle
 
         final ConcurrentMap<String, RegisteredEventListener<?>> eventsToBind = registries
                 .values().stream()
-                .flatMap(r -> r.getDomainEventListeners().get(DEFAULT_LISTENER).stream())
+                .flatMap(r -> r.getDomainEventListeners().get(DEFAULT_DOMAIN).stream())
                 .collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler),
                         ConcurrentHashMap::putAll);
 
         // event handlers and dynamic handlers
         final ConcurrentMap<String, RegisteredEventListener<?>> eventHandlers = registries
                 .values().stream()
-                .flatMap(r -> Stream.concat(r.getDomainEventListeners().get(DEFAULT_LISTENER).stream(), r.getDynamicEventHandlers().stream()))
+                .flatMap(r -> Stream.concat(r.getDomainEventListeners().get(DEFAULT_DOMAIN).stream(), r.getDynamicEventHandlers().stream()))
                 .collect(ConcurrentHashMap::new, (map, handler) -> map.put(handler.getPath(), handler),
                         ConcurrentHashMap::putAll);
 
diff --git a/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/RabbitDirectAsyncGateway.java b/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/RabbitDirectAsyncGateway.java
diff --git a/async/async-rabbit/src/test/java/org/reactivecommons/async/rabbit/listeners/ListenerReporterTestSuperClass.java b/async/async-rabbit/src/test/java/org/reactivecommons/async/rabbit/listeners/ListenerReporterTestSuperClass.java

Original file line number	Diff line number	Diff line change
`@@ -21,6 +21,7 @@ public static class DomainConnections {`
`21`	`21`	`private final ReactiveMessageListener listener;`
`22`	`22`	`private final ReactiveMessageSender sender;`
`23`	`23`	`private final HandlerResolver handlerResolver;`
	`24`	`+ private final ConnectionFactoryProvider provider;`
`24`	`25`	`@Setter`
`25`	`26`	`private DiscardNotifier discardNotifier;`
`26`	`27`	`}`
`@@ -62,4 +63,7 @@ public DiscardNotifier getDiscardNotifier(String domain) {`
`62`	`63`	`public HandlerResolver getHandlerResolver(String domain) {`
`63`	`64`	`return connections.get(domain).getHandlerResolver();`
`64`	`65`	`}`
	`66`	`+ public ConnectionFactoryProvider getProvider(String domain) {`
	`67`	`+ return connections.get(domain).getProvider();`
	`68`	`+ }`
`65`	`69`	`}`