fix(notifier): discard notifier for eda variant #112

fix(notifier): discard notifier for eda variant

Author: juancgalvis

.github/workflows/main.yml

@@ -44,7 +44,7 @@ jobs:
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
-        run: ./gradlew build test jacocoTestReport sonarqube --refresh-dependencies --no-daemon --continue -Denv.ci=true
+        run: ./gradlew build test jacocoTestReport sonar --refresh-dependencies --no-daemon --continue -Denv.ci=true
       - name: Execute build test jacocoTestReport pull request
         if: endsWith(github.REF, '/merge') == true
         env:

acceptance/async-tests/async-tests.gradle

@@ -3,4 +3,4 @@ dependencies {
     implementation 'org.springframework.boot:spring-boot-starter'
 }
 
-test.onlyIf { false }
+test.onlyIf { false }

async/async-commons-api/async-commons-api.gradle

@@ -7,5 +7,5 @@ dependencies {
     api project(':domain-events-api')
     compileOnly 'io.projectreactor:reactor-core'
     testImplementation 'io.projectreactor:reactor-test'
-    implementation "io.cloudevents:cloudevents-json-jackson:${cloudEventsVersion}"
-}
+    implementation 'io.cloudevents:cloudevents-json-jackson:4.0.1'
+}

async/async-commons/async-commons.gradle

@@ -8,9 +8,9 @@ dependencies {
     api project(':domain-events-api')
 
     compileOnly 'io.projectreactor:reactor-core'
-    api "io.projectreactor.rabbitmq:reactor-rabbitmq:${reactorRabbitVersion}"
+    api 'io.projectreactor.rabbitmq:reactor-rabbitmq:1.5.6'
     api 'com.fasterxml.jackson.core:jackson-databind'
-    implementation 'commons-io:commons-io:2.11.0'
+    implementation 'commons-io:commons-io:2.16.1'
 
     testImplementation 'io.projectreactor:reactor-test'
-}
+}

async/async-rabbit-starter-eda/async-commons-rabbit-starter-eda.gradle

@@ -13,4 +13,4 @@ dependencies {
 
     testImplementation 'io.projectreactor:reactor-test'
     testImplementation 'org.springframework.boot:spring-boot-starter-actuator'
-}
+}

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;
     }
 

async/async-rabbit-starter/async-commons-rabbit-starter.gradle

@@ -12,4 +12,4 @@ dependencies {
 
     testImplementation 'io.projectreactor:reactor-test'
     testImplementation 'org.springframework.boot:spring-boot-starter-actuator'
-}
+}

async/async-rabbit/async-rabbit.gradle

@@ -14,5 +14,5 @@ dependencies {
     api 'com.rabbitmq:amqp-client'
     api 'com.fasterxml.jackson.core:jackson-databind'
     testImplementation 'io.projectreactor:reactor-test'
-    implementation "io.cloudevents:cloudevents-json-jackson:${cloudEventsVersion}"
-}
+    implementation 'io.cloudevents:cloudevents-json-jackson:4.0.1'
+}