fix unit tests, update domainEvenListeners references, create shared constant for app events

juancgalvis · juancgalvis · commit ed325ec05b68 · 2023-06-27T08:27:29.000-05:00
diff --git a/async/async-commons-api/async-commons-api.gradle b/async/async-commons-api/async-commons-api.gradle
@@ -8,5 +8,4 @@ dependencies {
     compileOnly 'io.projectreactor:reactor-core'
     testImplementation 'io.projectreactor:reactor-test'
     implementation 'io.cloudevents:cloudevents-json-jackson:2.5.0'
-
 }
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,6 +23,7 @@
 @Getter
 @NoArgsConstructor(access = AccessLevel.PACKAGE)
 public class HandlerRegistry {
+    public static final String DEFAULT_LISTENER = "app";
     private final Map<String, List<RegisteredEventListener<?>>> domainEventListeners = new ConcurrentHashMap<>();
     private final List<RegisteredEventListener<?>> dynamicEventHandlers = new CopyOnWriteArrayList<>();
     private final List<RegisteredEventListener<?>> eventNotificationListener = new CopyOnWriteArrayList<>();
@@ -31,7 +32,9 @@ public class HandlerRegistry {
 
 
     public static HandlerRegistry register() {
-        return new HandlerRegistry();
+        HandlerRegistry instance = new HandlerRegistry();
+        instance.domainEventListeners.put(DEFAULT_LISTENER, new CopyOnWriteArrayList<>());
+        return instance;
     }
 
     public <T> HandlerRegistry listenDomainEvent(String domain, String eventName, EventHandler<T> handler, Class<T> eventClass) {
@@ -41,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("app", ignored -> new CopyOnWriteArrayList<>())
+        domainEventListeners.computeIfAbsent(DEFAULT_LISTENER, 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,6 +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;
 
 class HandlerRegistryTest {
     private final HandlerRegistry registry = HandlerRegistry.register();
@@ -27,7 +28,7 @@ void shouldListenEventWithTypeInferenceWhenClassInstanceIsUsed() {
 
         registry.listenEvent(name, eventHandler);
 
-        assertThat(registry.getEventListeners())
+        assertThat(registry.getDomainEventListeners().get(DEFAULT_LISTENER))
                 .anySatisfy(registered -> assertThat(registered)
                         .extracting(RegisteredEventListener::getPath, RegisteredEventListener::getInputClass, RegisteredEventListener::getHandler)
                         .containsExactly(name, SomeDataClass.class, eventHandler)).hasSize(1);
@@ -43,7 +44,7 @@ void shouldRegisterPatternEventHandlerWithTypeInference() {
         RegisteredEventListener<SomeDataClass> expectedRegisteredEventListener =
                 new RegisteredEventListener<>(eventNamePattern, eventHandler, SomeDataClass.class);
 
-        assertThat(registry.getEventListeners())
+        assertThat(registry.getDomainEventListeners().get(DEFAULT_LISTENER))
                 .anySatisfy(registeredEventListener -> assertThat(registeredEventListener)
                         .usingRecursiveComparison()
                         .isEqualTo(expectedRegisteredEventListener));
@@ -62,7 +63,7 @@ void shouldRegisterPatternEventHandler() {
         RegisteredEventListener<SomeDataClass> expectedRegisteredEventListener =
                 new RegisteredEventListener<>(eventNamePattern, eventHandler, SomeDataClass.class);
 
-        assertThat(registry.getEventListeners())
+        assertThat(registry.getDomainEventListeners().get(DEFAULT_LISTENER))
                 .anySatisfy(registeredEventListener -> assertThat(registeredEventListener)
                         .usingRecursiveComparison()
                         .isEqualTo(expectedRegisteredEventListener));
@@ -84,7 +85,7 @@ public void listenEvent() {
         EventHandler<SomeDataClass> handler = mock(EventHandler.class);
         registry.listenEvent(name, handler, SomeDataClass.class);
 
-        assertThat(registry.getEventListeners())
+        assertThat(registry.getDomainEventListeners().get(DEFAULT_LISTENER))
                 .anySatisfy(registered -> assertThat(registered)
                         .extracting(RegisteredEventListener::getPath, RegisteredEventListener::getInputClass, RegisteredEventListener::getHandler)
                         .containsExactly(name, SomeDataClass.class, handler)).hasSize(1);
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
@@ -16,6 +16,8 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 
+import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_LISTENER;
+
 @Configuration
 @Import(RabbitMqConfig.class)
 @RequiredArgsConstructor
@@ -25,12 +27,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("app"), props.getDirectMessagesExchangeName(), converter, meterRegistry);
+        return new RabbitDirectAsyncGateway(config, router, manager.getSender(DEFAULT_LISTENER), props.getDirectMessagesExchangeName(), converter, meterRegistry);
     }
 
     @Bean
     public ApplicationReplyListener msgListener(ReactiveReplyRouter router, BrokerConfig config, ConnectionManager manager) {
-        final ApplicationReplyListener replyListener = new ApplicationReplyListener(router, manager.getListener("app"), props.getReplyQueue());
+        final ApplicationReplyListener replyListener = new ApplicationReplyListener(router, manager.getListener(DEFAULT_LISTENER), 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,6 +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 reactor.rabbitmq.ExchangeSpecification.exchange;
 
 @Configuration
@@ -21,12 +22,11 @@ public class EventBusConfig {
     @Bean // app connection
     public DomainEventBus domainEventBus(ConnectionManager manager, BrokerConfigProps props, BrokerConfig config,
                                          ObjectMapperSupplier objectMapperSupplier) {
-        String domain = "app";
-        ReactiveMessageSender sender = manager.getSender(domain);
+        ReactiveMessageSender sender = manager.getSender(DEFAULT_LISTENER);
         final String exchangeName = props.getDomainEventsExchangeName();
         sender.getTopologyCreator().declare(exchange(exchangeName).durable(true).type("topic")).subscribe();
         DomainEventBus domainEventBus = new RabbitDomainEventBus(sender, exchangeName, config);
-        manager.setDiscardNotifier(domain, createDiscardNotifier(domainEventBus, objectMapperSupplier));
+        manager.setDiscardNotifier(DEFAULT_LISTENER, 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,6 +12,8 @@
 
 import java.util.concurrent.atomic.AtomicReference;
 
+import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_LISTENER;
+
 @Configuration
 @RequiredArgsConstructor
 @Import(RabbitMqConfig.class)
@@ -38,7 +40,7 @@ public ApplicationEventListener eventListener(MessageConverter messageConverter,
                     manager.getDiscardNotifier(domain),
                     errorReporter,
                     appName);
-            if ("app".equals(domain)) {
+            if (DEFAULT_LISTENER.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,14 +14,16 @@
 import java.util.concurrent.ConcurrentMap;
 import java.util.stream.Stream;
 
+import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_LISTENER;
+
 @NoArgsConstructor(access = AccessLevel.PRIVATE)
 public class HandlerResolverBuilder {
 
     public static HandlerResolver buildResolver(String domain,
                                                 Map<String, HandlerRegistry> registries,
                                                 final DefaultCommandHandler defaultCommandHandler) {
 
-        if ("app".equals(domain)) {
+        if (DEFAULT_LISTENER.equals(domain)) {
             final ConcurrentMap<String, RegisteredQueryHandler<?, ?>> queryHandlers = registries
                     .values().stream()
                     .flatMap(r -> r.getHandlers().stream())
@@ -79,7 +81,7 @@ private static ConcurrentMap<String, RegisteredEventListener<?>> getEventHandler
     }
 
     private static Stream<RegisteredEventListener<?>> getDynamics(String domain, HandlerRegistry r) {
-        if ("app".equals(domain)) {
+        if (DEFAULT_LISTENER.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/RabbitMqConfig.java b/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/RabbitMqConfig.java
@@ -52,6 +52,8 @@
 import java.util.Map;
 import java.util.logging.Level;
 
+import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_LISTENER;
+
 @Log
 @Configuration
 @RequiredArgsConstructor
@@ -193,8 +195,8 @@ Mono<Connection> createConnectionMono(ConnectionFactory factory, String connecti
 
     @Bean
     public DynamicRegistry dynamicRegistry(ConnectionManager connectionManager, IBrokerConfigProps props) {
-        return new DynamicRegistryImp(connectionManager.getHandlerResolver("app"),
-                connectionManager.getListener("app").getTopologyCreator(), props);
+        return new DynamicRegistryImp(connectionManager.getHandlerResolver(DEFAULT_LISTENER),
+                connectionManager.getListener(DEFAULT_LISTENER).getTopologyCreator(), props);
     }
 
     @Bean
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
@@ -4,17 +4,22 @@
 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.api.HandlerRegistry;
 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.rabbit.listeners.ApplicationEventListener;
 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;
 
@@ -27,12 +32,13 @@ class EventListenersConfigTest {
     private final AsyncProps props = new AsyncProps();
     private final EventListenersConfig config = new EventListenersConfig(props);
     private final ReactiveMessageListener listener = mock(ReactiveMessageListener.class);
+    private final ReactiveMessageSender sender = mock(ReactiveMessageSender.class);
     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 ConnectionManager connectionManager;
 
     @BeforeEach
     public void init() {
@@ -47,15 +53,15 @@ 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);
+        connectionManager = new ConnectionManager();
+        connectionManager.addDomain(HandlerRegistry.DEFAULT_LISTENER, listener, sender, handlerResolver);
     }
 
     @Test
     void eventListener() {
         final ApplicationEventListener eventListener = config.eventListener(
-                handlerResolver,
                 messageConverter,
-                listener,
-                discardNotifier,
+                connectionManager,
                 customReporter
         );
 
diff --git a/async/async-rabbit-starter-eda/src/test/java/org/reactivecommons/async/rabbit/config/RabbitMqConfigTest.java b/async/async-rabbit-starter-eda/src/test/java/org/reactivecommons/async/rabbit/config/RabbitMqConfigTest.java
@@ -22,7 +22,7 @@
 
 class RabbitMqConfigTest {
 
-    RabbitMqConfig config = new RabbitMqConfig(null);
+    RabbitMqConfig config = new RabbitMqConfig();
 
     @Test
     void retryInitialConnection() throws IOException, TimeoutException {
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,6 +62,7 @@
 import java.util.logging.Level;
 import java.util.stream.Stream;
 
+import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_LISTENER;
 import static reactor.rabbitmq.ExchangeSpecification.exchange;
 
 @Log
@@ -213,14 +214,14 @@ public HandlerResolver resolver(ApplicationContext context, DefaultCommandHandle
 
         final ConcurrentMap<String, RegisteredEventListener<?>> eventsToBind = registries
                 .values().stream()
-                .flatMap(r -> r.getEventListeners().stream())
+                .flatMap(r -> r.getDomainEventListeners().get(DEFAULT_LISTENER).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.getEventListeners().stream(), r.getDynamicEventHandlers().stream()))
+                .flatMap(r -> Stream.concat(r.getDomainEventListeners().get(DEFAULT_LISTENER).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/converters/json/JacksonCloudEventMessageConverter.java b/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/converters/json/JacksonCloudEventMessageConverter.java
@@ -21,6 +21,7 @@
 
 public class JacksonCloudEventMessageConverter implements MessageConverter {
     private static final String CONTENT_TYPE = "application/json";
+    public static final String FAILED_TO_CONVERT_MESSAGE_CONTENT = "Failed to convert Message content";
 
     private final ObjectMapper objectMapper;
 
@@ -36,7 +37,7 @@ public <T> AsyncQuery<T> readAsyncQuery(Message message, Class<T> bodyClass) {
             T value = extractData(bodyClass, asyncQueryJson.getQueryData());
             return new AsyncQuery<>(asyncQueryJson.getResource(), value);
         } catch (IOException e) {
-            throw new MessageConversionException("Failed to convert Message content", e);
+            throw new MessageConversionException(FAILED_TO_CONVERT_MESSAGE_CONTENT, e);
         }
     }
 
@@ -49,7 +50,7 @@ public <T> DomainEvent<T> readDomainEvent(Message message, Class<T> bodyClass) {
 
             return new DomainEvent<>(domainEventJson.getName(), domainEventJson.getEventId(), value);
         } catch (IOException e) {
-            throw new MessageConversionException("Failed to convert Message content", e);
+            throw new MessageConversionException(FAILED_TO_CONVERT_MESSAGE_CONTENT, e);
         }
     }
 
@@ -60,7 +61,7 @@ public <T> Command<T> readCommand(Message message, Class<T> bodyClass) {
             T value = extractData(bodyClass, commandJson.getData());
             return new Command<>(commandJson.getName(), commandJson.getCommandId(), value);
         } catch (IOException e) {
-            throw new MessageConversionException("Failed to convert Message content", e);
+            throw new MessageConversionException(FAILED_TO_CONVERT_MESSAGE_CONTENT, e);
         }
     }
 
@@ -77,7 +78,7 @@ public <T> T readValue(Message message, Class<T> valueClass) {
             }
             return objectMapper.readValue(message.getBody(), valueClass);
         } catch (IOException e) {
-            throw new MessageConversionException("Failed to convert Message content", e);
+            throw new MessageConversionException(FAILED_TO_CONVERT_MESSAGE_CONTENT, e);
         }
     }
 
@@ -133,35 +134,29 @@ public Message queryToMessage(AsyncQuery<CloudEvent> object) {
     }
     @Override
     public Message toMessage(Object object) {
-        byte[] bytes;
         if(object instanceof DomainEvent
-                && ((DomainEvent) object).getData() instanceof CloudEvent){
-
+                && ((DomainEvent<?>) object).getData() instanceof CloudEvent){
            return eventToMessage((DomainEvent) object);
 
         }
         if(object instanceof Command
-                && ((Command) object).getData() instanceof CloudEvent){
-
+                && ((Command<?>) object).getData() instanceof CloudEvent){
             return commandToMessage((Command) object);
         }
         if(object instanceof AsyncQuery
-                && ((AsyncQuery) object).getQueryData() instanceof CloudEvent){
-
+                && ((AsyncQuery<?>) object).getQueryData() instanceof CloudEvent){
             return queryToMessage((AsyncQuery) object);
         }
-
         return getRabbitMessage(object);
     }
 
     private RabbitMessage getRabbitMessage(Object object) {
         byte[] bytes;
         try {
-
             String jsonString = this.objectMapper.writeValueAsString(object);
             bytes = jsonString.getBytes(StandardCharsets.UTF_8);
         } catch (IOException e) {
-            throw new MessageConversionException("Failed to convert Message content", e);
+            throw new MessageConversionException(FAILED_TO_CONVERT_MESSAGE_CONTENT, e);
         }
         RabbitMessage.RabbitMessageProperties props = new RabbitMessage.RabbitMessageProperties();
         props.setContentType(CONTENT_TYPE);
diff --git a/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/converters/json/JacksonMessageConverter.java b/async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/converters/json/JacksonMessageConverter.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
`@@ -8,5 +8,4 @@ dependencies {`
`8`	`8`	`compileOnly 'io.projectreactor:reactor-core'`
`9`	`9`	`testImplementation 'io.projectreactor:reactor-test'`
`10`	`10`	`implementation 'io.cloudevents:cloudevents-json-jackson:2.5.0'`
`11`		`-`
`12`	`11`	`}`