fix(notifier): discard notifier for eda variant

Author: juancgalvis

async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/ConnectionManager.java

@@ -32,8 +32,8 @@ public void forListener(BiConsumer<String, ReactiveMessageListener> consumer) {
         connections.forEach((key, conn) -> consumer.accept(key, conn.getListener()));
     }
 
-    public void setDiscardNotifier(String domain, DiscardNotifier discardNotifier) {
-        getChecked(domain).setDiscardNotifier(discardNotifier);
+    public void setDiscardNotifierForAll(DiscardNotifier discardNotifier) {
+        connections.forEach((key, conn) -> conn.setDiscardNotifier(discardNotifier));
     }
 
     public ConnectionManager addDomain(String domain, ReactiveMessageListener listener, ReactiveMessageSender sender,

async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/EventBusConfig.java

@@ -1,10 +1,7 @@
 package org.reactivecommons.async.rabbit.config;
 
 import org.reactivecommons.api.domain.DomainEventBus;
-import org.reactivecommons.async.commons.DiscardNotifier;
 import org.reactivecommons.async.commons.config.BrokerConfig;
-import org.reactivecommons.async.commons.converters.json.ObjectMapperSupplier;
-import org.reactivecommons.async.rabbit.RabbitDiscardNotifier;
 import org.reactivecommons.async.rabbit.RabbitDomainEventBus;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageSender;
 import org.reactivecommons.async.rabbit.config.props.AsyncProps;
@@ -22,19 +19,13 @@ public class EventBusConfig {
 
     @Bean // app connection
     public DomainEventBus domainEventBus(ConnectionManager manager, BrokerConfig config,
-                                         AsyncPropsDomain asyncPropsDomain, ObjectMapperSupplier objectMapperSupplier) {
+                                         AsyncPropsDomain asyncPropsDomain) {
         ReactiveMessageSender sender = manager.getSender(DEFAULT_DOMAIN);
         AsyncProps asyncProps = asyncPropsDomain.getProps(DEFAULT_DOMAIN);
         final String exchangeName = asyncProps.getBrokerConfigProps().getDomainEventsExchangeName();
         if (asyncProps.getCreateTopology()) {
             sender.getTopologyCreator().declare(exchange(exchangeName).durable(true).type("topic")).subscribe();
         }
-        DomainEventBus domainEventBus = new RabbitDomainEventBus(sender, exchangeName, config);
-        manager.setDiscardNotifier(DEFAULT_DOMAIN, createDiscardNotifier(domainEventBus, objectMapperSupplier));
-        return domainEventBus;
-    }
-
-    private DiscardNotifier createDiscardNotifier(DomainEventBus domainEventBus, ObjectMapperSupplier objectMapperSupplier) {
-        return new RabbitDiscardNotifier(domainEventBus, objectMapperSupplier.get());
+        return new RabbitDomainEventBus(sender, exchangeName, config);
     }
 }

async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/RabbitMqConfig.java

@@ -7,11 +7,13 @@
 import lombok.extern.java.Log;
 import org.reactivecommons.api.domain.Command;
 import org.reactivecommons.api.domain.DomainEvent;
+import org.reactivecommons.api.domain.DomainEventBus;
 import org.reactivecommons.async.api.AsyncQuery;
 import org.reactivecommons.async.api.DefaultCommandHandler;
 import org.reactivecommons.async.api.DefaultQueryHandler;
 import org.reactivecommons.async.api.DynamicRegistry;
 import org.reactivecommons.async.api.HandlerRegistry;
+import org.reactivecommons.async.commons.DiscardNotifier;
 import org.reactivecommons.async.commons.communications.Message;
 import org.reactivecommons.async.commons.config.BrokerConfig;
 import org.reactivecommons.async.commons.config.IBrokerConfigProps;
@@ -21,6 +23,8 @@
 import org.reactivecommons.async.commons.ext.CustomReporter;
 import org.reactivecommons.async.rabbit.DynamicRegistryImp;
 import org.reactivecommons.async.rabbit.HandlerResolver;
+import org.reactivecommons.async.rabbit.RabbitDiscardNotifier;
+import org.reactivecommons.async.rabbit.RabbitDomainEventBus;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageListener;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageSender;
 import org.reactivecommons.async.rabbit.communications.TopologyCreator;
@@ -67,14 +71,20 @@ public class RabbitMqConfig {
 
 
     @Bean
-    public ConnectionManager buildConnectionManager(AsyncPropsDomain props, MessageConverter converter) {
+    public ConnectionManager buildConnectionManager(AsyncPropsDomain props, MessageConverter converter,
+                                                    BrokerConfig brokerConfig, ObjectMapperSupplier objectMapperSupplier) {
         ConnectionManager connectionManager = new ConnectionManager();
         props.forEach((domain, properties) -> {
             ConnectionFactoryProvider provider = createConnectionFactoryProvider(properties.getConnectionProperties());
             ReactiveMessageSender sender = createMessageSender(provider, properties, converter);
             ReactiveMessageListener listener = createMessageListener(provider, properties);
             connectionManager.addDomain(domain, listener, sender, provider);
         });
+        ReactiveMessageSender appDomainSender = connectionManager.getSender(DEFAULT_DOMAIN);
+        DomainEventBus appDomainEventBus = new RabbitDomainEventBus(appDomainSender, props.getProps(DEFAULT_DOMAIN)
+                .getBrokerConfigProps().getDomainEventsExchangeName(), brokerConfig);
+        DiscardNotifier notifier = new RabbitDiscardNotifier(appDomainEventBus, objectMapperSupplier.get());
+        connectionManager.setDiscardNotifierForAll(notifier);
         return connectionManager;
     }