update commands and queries to listen from default app doman

juancgalvis · juancgalvis · commit 4df5c3ff36d0 · 2023-06-28T08:24:21.000-05:00
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
@@ -1,18 +1,17 @@
 package org.reactivecommons.async.rabbit.config;
 
 import lombok.RequiredArgsConstructor;
-import org.reactivecommons.async.commons.DiscardNotifier;
-import org.reactivecommons.async.rabbit.HandlerResolver;
-import org.reactivecommons.async.rabbit.communications.ReactiveMessageListener;
-import org.reactivecommons.async.rabbit.config.props.AsyncProps;
 import org.reactivecommons.async.commons.converters.MessageConverter;
 import org.reactivecommons.async.commons.ext.CustomReporter;
+import org.reactivecommons.async.rabbit.config.props.AsyncProps;
 import org.reactivecommons.async.rabbit.listeners.ApplicationCommandListener;
 import org.springframework.beans.factory.annotation.Value;
 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_DOMAIN;
+
 @Configuration
 @RequiredArgsConstructor
 @Import(RabbitMqConfig.class)
@@ -24,13 +23,12 @@ public class CommandListenersConfig {
     private final AsyncProps asyncProps;
 
     @Bean
-    public ApplicationCommandListener applicationCommandListener(ReactiveMessageListener listener,
-                                                                 HandlerResolver resolver, MessageConverter converter,
-                                                                 DiscardNotifier discardNotifier,
+    public ApplicationCommandListener applicationCommandListener(ConnectionManager manager,
+                                                                 MessageConverter converter,
                                                                  CustomReporter errorReporter) {
-        ApplicationCommandListener commandListener = new ApplicationCommandListener(listener, appName, resolver,
+        ApplicationCommandListener commandListener = new ApplicationCommandListener(manager.getListener(DEFAULT_DOMAIN), appName, manager.getHandlerResolver(DEFAULT_DOMAIN),
                 asyncProps.getDirect().getExchange(), converter, asyncProps.getWithDLQRetry(), asyncProps.getMaxRetries(),
-                asyncProps.getRetryDelay(), asyncProps.getDirect().getMaxLengthBytes(), discardNotifier, errorReporter);
+                asyncProps.getRetryDelay(), asyncProps.getDirect().getMaxLengthBytes(), manager.getDiscardNotifier(DEFAULT_DOMAIN), errorReporter);
 
         commandListener.startListener();
 
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
@@ -29,7 +29,7 @@ public RabbitDirectAsyncGateway rabbitDirectAsyncGateway(BrokerConfig config, Re
         return new RabbitEDADirectAsyncGateway(config, router, manager, props.getDirectMessagesExchangeName(), converter, meterRegistry);
     }
 
-    @Bean
+    @Bean // TODO: Listen replies from n domains if enabled in config
     public ApplicationReplyListener msgListener(ReactiveReplyRouter router, BrokerConfig config, ConnectionManager manager) {
         final ApplicationReplyListener replyListener = new ApplicationReplyListener(router, manager.getListener(DEFAULT_DOMAIN), props.getReplyQueue());
         replyListener.startListening(config.getRoutingKey());
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
@@ -1,18 +1,17 @@
 package org.reactivecommons.async.rabbit.config;
 
 import lombok.RequiredArgsConstructor;
-import org.reactivecommons.async.commons.DiscardNotifier;
-import org.reactivecommons.async.rabbit.HandlerResolver;
-import org.reactivecommons.async.rabbit.communications.ReactiveMessageListener;
-import org.reactivecommons.async.rabbit.config.props.AsyncProps;
 import org.reactivecommons.async.commons.converters.MessageConverter;
 import org.reactivecommons.async.commons.ext.CustomReporter;
+import org.reactivecommons.async.rabbit.config.props.AsyncProps;
 import org.reactivecommons.async.rabbit.listeners.ApplicationNotificationListener;
 import org.springframework.beans.factory.annotation.Value;
 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_DOMAIN;
+
 @Configuration
 @RequiredArgsConstructor
 @Import(RabbitMqConfig.class)
@@ -24,15 +23,16 @@ public class NotificacionListenersConfig {
     private final AsyncProps asyncProps;
 
     @Bean
-    public ApplicationNotificationListener eventNotificationListener(HandlerResolver resolver, MessageConverter messageConverter,
-                                                                     ReactiveMessageListener receiver, DiscardNotifier discardNotifier, CustomReporter errorReporter) {
+    public ApplicationNotificationListener eventNotificationListener(ConnectionManager manager,
+                                                                     MessageConverter messageConverter,
+                                                                     CustomReporter errorReporter) {
         final ApplicationNotificationListener listener = new ApplicationNotificationListener(
-                receiver,
+                manager.getListener(DEFAULT_DOMAIN),
                 asyncProps.getDomain().getEvents().getExchange(),
                 asyncProps.getNotificationProps().getQueueName(appName),
-                resolver,
+                manager.getHandlerResolver(DEFAULT_DOMAIN),
                 messageConverter,
-                discardNotifier,
+                manager.getDiscardNotifier(DEFAULT_DOMAIN),
                 errorReporter);
         listener.startListener();
         return listener;
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
@@ -1,19 +1,17 @@
 package org.reactivecommons.async.rabbit.config;
 
 import lombok.RequiredArgsConstructor;
-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;
-import org.reactivecommons.async.rabbit.config.props.AsyncProps;
 import org.reactivecommons.async.commons.converters.MessageConverter;
 import org.reactivecommons.async.commons.ext.CustomReporter;
+import org.reactivecommons.async.rabbit.config.props.AsyncProps;
 import org.reactivecommons.async.rabbit.listeners.ApplicationQueryListener;
 import org.springframework.beans.factory.annotation.Value;
 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_DOMAIN;
+
 @Configuration
 @RequiredArgsConstructor
 @Import(RabbitMqConfig.class)
@@ -25,15 +23,14 @@ public class QueryListenerConfig {
     private final AsyncProps asyncProps;
 
     @Bean
-    public ApplicationQueryListener queryListener(MessageConverter converter, HandlerResolver resolver,
-                                                  ReactiveMessageSender sender, ReactiveMessageListener rlistener,
-                                                  DiscardNotifier discardNotifier,
+    public ApplicationQueryListener queryListener(MessageConverter converter,
+                                                  ConnectionManager manager,
                                                   CustomReporter errorReporter) {
-        final ApplicationQueryListener listener = new ApplicationQueryListener(rlistener,
-                appName + ".query", resolver, sender, asyncProps.getDirect().getExchange(), converter,
+        final ApplicationQueryListener listener = new ApplicationQueryListener(manager.getListener(DEFAULT_DOMAIN),
+                appName + ".query", manager.getHandlerResolver(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(),  discardNotifier, errorReporter);
+                asyncProps.getRetryDelay(), asyncProps.getGlobal().getMaxLengthBytes(),
+                asyncProps.getDirect().isDiscardTimeoutQueries(), manager.getDiscardNotifier(DEFAULT_DOMAIN), errorReporter);
 
         listener.startListener();
 
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
@@ -6,7 +6,6 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.junit.jupiter.MockitoExtension;
-import org.reactivecommons.async.commons.DiscardNotifier;
 import org.reactivecommons.async.commons.converters.MessageConverter;
 import org.reactivecommons.async.commons.ext.CustomReporter;
 import org.reactivecommons.async.rabbit.HandlerResolver;
@@ -26,6 +25,7 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
+import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_DOMAIN;
 
 @ExtendWith(MockitoExtension.class)
 class CommandListenersConfigTest {
@@ -37,9 +37,9 @@ class CommandListenersConfigTest {
     private final TopologyCreator creator = mock(TopologyCreator.class);
     private final HandlerResolver handlerResolver = mock(HandlerResolver.class);
     private final MessageConverter messageConverter = mock(MessageConverter.class);
-    private final DiscardNotifier discardNotifier = mock(DiscardNotifier.class);
     private final CustomReporter customReporter = mock(CustomReporter.class);
     private final Receiver receiver = mock(Receiver.class);
+    private final ConnectionManager manager = new ConnectionManager();
 
     @BeforeEach
     public void init() throws NoSuchFieldException, IllegalAccessException {
@@ -53,17 +53,12 @@ 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);
     }
 
     @Test
     void applicationCommandListener() {
-        final ApplicationCommandListener commandListener = config.applicationCommandListener(
-                listener,
-                handlerResolver,
-                messageConverter,
-                discardNotifier,
-                customReporter
-        );
+        final ApplicationCommandListener commandListener = config.applicationCommandListener(manager, messageConverter, customReporter);
         Assertions.assertThat(commandListener).isNotNull();
     }
 }
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
@@ -5,22 +5,27 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.reactivecommons.async.commons.DiscardNotifier;
+import org.reactivecommons.async.commons.converters.MessageConverter;
+import org.reactivecommons.async.commons.ext.CustomReporter;
 import org.reactivecommons.async.rabbit.HandlerResolver;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageListener;
 import org.reactivecommons.async.rabbit.communications.TopologyCreator;
 import org.reactivecommons.async.rabbit.config.props.AsyncProps;
-import org.reactivecommons.async.commons.converters.MessageConverter;
-import org.reactivecommons.async.commons.ext.CustomReporter;
 import org.reactivecommons.async.rabbit.listeners.ApplicationNotificationListener;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.rabbitmq.*;
+import reactor.rabbitmq.BindingSpecification;
+import reactor.rabbitmq.ConsumeOptions;
+import reactor.rabbitmq.ExchangeSpecification;
+import reactor.rabbitmq.QueueSpecification;
+import reactor.rabbitmq.Receiver;
 
 import java.util.Collections;
 
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
+import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_DOMAIN;
 
 class NotificacionListenersConfigTest {
 
@@ -33,6 +38,7 @@ class NotificacionListenersConfigTest {
     private final DiscardNotifier discardNotifier = mock(DiscardNotifier.class);
     private final CustomReporter customReporter = mock(CustomReporter.class);
     private final Receiver receiver = mock(Receiver.class);
+    private final ConnectionManager manager = new ConnectionManager();
 
     @BeforeEach
     public void init() {
@@ -47,12 +53,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, null, handlerResolver);
     }
 
     @Test
     void eventNotificationListener() {
-        final ApplicationNotificationListener applicationEventListener = config.
-                eventNotificationListener(handlerResolver, messageConverter, listener, discardNotifier, customReporter);
+        final ApplicationNotificationListener applicationEventListener =
+                config.eventNotificationListener(manager, 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
@@ -4,24 +4,28 @@
 import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.reactivecommons.async.commons.DiscardNotifier;
+import org.reactivecommons.async.commons.converters.MessageConverter;
+import org.reactivecommons.async.commons.ext.CustomReporter;
 import org.reactivecommons.async.rabbit.HandlerResolver;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageListener;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageSender;
 import org.reactivecommons.async.rabbit.communications.TopologyCreator;
 import org.reactivecommons.async.rabbit.config.props.AsyncProps;
-import org.reactivecommons.async.commons.converters.MessageConverter;
-import org.reactivecommons.async.commons.ext.CustomReporter;
 import org.reactivecommons.async.rabbit.listeners.ApplicationQueryListener;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.rabbitmq.*;
+import reactor.rabbitmq.BindingSpecification;
+import reactor.rabbitmq.ConsumeOptions;
+import reactor.rabbitmq.ExchangeSpecification;
+import reactor.rabbitmq.QueueSpecification;
+import reactor.rabbitmq.Receiver;
 
 import java.util.Collections;
 
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
+import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_DOMAIN;
 
 class QueryListenerConfigTest {
 
@@ -31,10 +35,10 @@ class QueryListenerConfigTest {
     private final TopologyCreator creator = mock(TopologyCreator.class);
     private final HandlerResolver handlerResolver = mock(HandlerResolver.class);
     private final MessageConverter messageConverter = mock(MessageConverter.class);
-    private final DiscardNotifier discardNotifier = mock(DiscardNotifier.class);
     private final CustomReporter customReporter = mock(CustomReporter.class);
     private final Receiver receiver = mock(Receiver.class);
     private final ReactiveMessageSender sender = mock(ReactiveMessageSender.class);
+    private final ConnectionManager manager = new ConnectionManager();
 
     @BeforeEach
     public void init() {
@@ -49,11 +53,12 @@ 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);
     }
 
     @Test
     void queryListener() {
-        final ApplicationQueryListener queryListener = config.queryListener(messageConverter, handlerResolver, sender, listener, discardNotifier, customReporter);
+        final ApplicationQueryListener queryListener = config.queryListener(messageConverter, manager, customReporter);
         Assertions.assertThat(queryListener).isNotNull();
     }
 }
diff --git a/gradle.properties b/gradle.properties
@@ -2,4 +2,4 @@ version=2.0.0
 springBootVersion=3.0.2
 reactorRabbitVersion=1.5.5
 gradleVersionsVersion=0.36.0
-toPublish=async-commons,async-commons-api,async-commons-rabbit-standalone,async-commons-rabbit-starter,domain-events-api,async-rabbit
+toPublish=async-commons,async-commons-api,async-commons-rabbit-standalone,async-commons-rabbit-starter,async-commons-rabbit-starter-eda,domain-events-api,async-rabbit

Original file line number	Diff line number	Diff line change
`@@ -29,7 +29,7 @@ public RabbitDirectAsyncGateway rabbitDirectAsyncGateway(BrokerConfig config, Re`
`29`	`29`	`return new RabbitEDADirectAsyncGateway(config, router, manager, props.getDirectMessagesExchangeName(), converter, meterRegistry);`
`30`	`30`	`}`
`31`	`31`
`32`		`- @Bean`
	`32`	`+ @Bean // TODO: Listen replies from n domains if enabled in config`
`33`	`33`	`public ApplicationReplyListener msgListener(ReactiveReplyRouter router, BrokerConfig config, ConnectionManager manager) {`
`34`	`34`	`final ApplicationReplyListener replyListener = new ApplicationReplyListener(router, manager.getListener(DEFAULT_DOMAIN), props.getReplyQueue());`
`35`	`35`	`replyListener.startListening(config.getRoutingKey());`