Implement Discart notification by events.

Author: Daniel Bustamante Ospina

acceptance/async-tests/src/test/java/org/reactivecommons/test/DynamicRegistryTest.java

@@ -2,15 +2,10 @@
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.mockito.Mockito;
 import org.reactivecommons.api.domain.DomainEvent;
 import org.reactivecommons.api.domain.DomainEventBus;
-import org.reactivecommons.async.api.AsyncQuery;
-import org.reactivecommons.async.api.DirectAsyncGateway;
 import org.reactivecommons.async.api.DynamicRegistry;
-import org.reactivecommons.async.api.HandlerRegistry;
 import org.reactivecommons.async.api.handlers.EventHandler;
-import org.reactivecommons.async.impl.config.annotations.EnableDirectAsyncGateway;
 import org.reactivecommons.async.impl.config.annotations.EnableDomainEventBus;
 import org.reactivecommons.async.impl.config.annotations.EnableMessageListeners;
 import org.reactivestreams.Publisher;
@@ -19,9 +14,7 @@
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.context.annotation.Bean;
 import org.springframework.test.context.junit4.SpringRunner;
-import reactor.core.publisher.Mono;
 import reactor.core.publisher.UnicastProcessor;
 import reactor.test.StepVerifier;
 

acceptance/async-tests/src/test/java/org/reactivecommons/test/SimpleDirectCommunicationTest.java

@@ -16,15 +16,12 @@
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.context.annotation.Bean;
 import org.springframework.test.context.junit4.SpringRunner;
-import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.core.publisher.SignalType;
 import reactor.core.publisher.UnicastProcessor;
 import reactor.test.StepVerifier;
 
 import java.time.Duration;
 import java.util.concurrent.ThreadLocalRandom;
-import java.util.logging.Level;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static reactor.core.publisher.Mono.*;

acceptance/async-tests/src/test/java/org/reactivecommons/test/SimpleEventNotificationTest.java

@@ -18,7 +18,6 @@
 import reactor.test.StepVerifier;
 
 import java.util.concurrent.ThreadLocalRandom;
-import java.util.concurrent.TimeUnit;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static reactor.core.publisher.Mono.*;

acceptance/async-tests/src/test/java/org/reactivecommons/test/perf/BlockingCommandHandlePerfTest.java

@@ -1,6 +1,5 @@
 package org.reactivecommons.test.perf;
 
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.reactivecommons.api.domain.Command;
@@ -17,9 +16,7 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.test.context.junit4.SpringRunner;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
 import reactor.core.publisher.UnicastProcessor;
-import reactor.core.scheduler.Schedulers;
 import reactor.test.StepVerifier;
 
 import java.time.Duration;

acceptance/async-tests/src/test/java/org/reactivecommons/test/perf/ParallelOnBlockingInSubscriptionTimeTest.java

@@ -21,14 +21,12 @@
 import reactor.test.StepVerifier;
 
 import java.time.Duration;
-import java.util.List;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 
 import static java.lang.System.out;
 import static org.assertj.core.api.Assertions.assertThat;
 import static reactor.core.publisher.Mono.empty;
-import static reactor.core.publisher.Mono.fromRunnable;
 
 @SpringBootTest
 @RunWith(SpringRunner.class)

acceptance/async-tests/src/test/java/org/reactivecommons/test/perf/SimpleCommandHandlePerfTest.java

@@ -13,11 +13,8 @@
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.ImportResource;
-import org.springframework.context.annotation.PropertySource;
 import org.springframework.test.context.junit4.SpringRunner;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.UnicastProcessor;

async/async-commons-api/src/main/java/org/reactivecommons/async/api/HandlerRegistry.java

@@ -46,7 +46,6 @@ public <T> HandlerRegistry handleCommand(String commandName, CommandHandler<T> f
         return this;
     }
 
-    @SuppressWarnings("unchecked")
     public <T, R> HandlerRegistry serveQuery(String resource, QueryHandler<T, R> handler){
         return serveQuery(resource, handler, inferGenericParameterType(handler));
     }

async/async-commons-standalone/src/main/java/org/reactivecommons/async/impl/config/RabbitMqConfig.java

@@ -3,7 +3,6 @@
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
 import lombok.extern.java.Log;
-import org.reactivecommons.async.impl.communications.ReactiveMessageListener;
 import org.reactivecommons.async.impl.communications.ReactiveMessageSender;
 import org.reactivecommons.async.impl.communications.TopologyCreator;
 import org.reactivecommons.async.impl.converters.json.JacksonMessageConverter;

async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/ReplyCommandSender.java

@@ -1,6 +1,5 @@
 package org.reactivecommons.async.impl;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import org.reactivecommons.async.api.AsyncQuery;
 import org.reactivecommons.async.api.DirectAsyncGateway;
 import org.springframework.beans.factory.annotation.Autowired;

async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/MessageListenersConfig.java

@@ -4,16 +4,14 @@
 import org.reactivecommons.async.api.DefaultQueryHandler;
 import org.reactivecommons.async.api.DynamicRegistry;
 import org.reactivecommons.async.api.HandlerRegistry;
-import org.reactivecommons.async.api.handlers.QueryHandler;
 import org.reactivecommons.async.api.handlers.registered.RegisteredCommandHandler;
 import org.reactivecommons.async.api.handlers.registered.RegisteredEventListener;
 import org.reactivecommons.async.api.handlers.registered.RegisteredQueryHandler;
+import org.reactivecommons.async.impl.DiscardNotifier;
 import org.reactivecommons.async.impl.DynamicRegistryImp;
 import org.reactivecommons.async.impl.HandlerResolver;
 import org.reactivecommons.async.impl.communications.ReactiveMessageListener;
 import org.reactivecommons.async.impl.communications.ReactiveMessageSender;
-import org.reactivecommons.async.impl.communications.TopologyCreator;
-import org.reactivecommons.async.impl.config.props.BrokerConfigProps;
 import org.reactivecommons.async.impl.converters.MessageConverter;
 import org.reactivecommons.async.impl.listeners.ApplicationCommandListener;
 import org.reactivecommons.async.impl.listeners.ApplicationEventListener;
@@ -54,22 +52,22 @@ public class MessageListenersConfig {
     private boolean withDLQRetry;
 
     @Bean //TODO: move to own config (QueryListenerConfig)
-    public ApplicationEventListener eventListener(HandlerResolver resolver, MessageConverter messageConverter, ReactiveMessageListener receiver) throws Exception {
-        final ApplicationEventListener listener = new ApplicationEventListener(receiver, appName + ".subsEvents", resolver, domainEventsExchangeName, messageConverter, withDLQRetry, maxRetries, retryDelay);
+    public ApplicationEventListener eventListener(HandlerResolver resolver, MessageConverter messageConverter, ReactiveMessageListener receiver, DiscardNotifier discardNotifier) throws Exception {
+        final ApplicationEventListener listener = new ApplicationEventListener(receiver, appName + ".subsEvents", resolver, domainEventsExchangeName, messageConverter, withDLQRetry, maxRetries, retryDelay, discardNotifier);
         listener.startListener();
         return listener;
     }
 
     @Bean //TODO: move to own config (QueryListenerConfig)
-    public ApplicationQueryListener queryListener(MessageConverter converter, HandlerResolver resolver, ReactiveMessageSender sender, ReactiveMessageListener rlistener) throws Exception {
-        final ApplicationQueryListener listener = new ApplicationQueryListener(rlistener, appName+".query", resolver, sender, directMessagesExchangeName, converter, "globalReply", withDLQRetry, maxRetries, retryDelay);
+    public ApplicationQueryListener queryListener(MessageConverter converter, HandlerResolver resolver, ReactiveMessageSender sender, ReactiveMessageListener rlistener, DiscardNotifier discardNotifier) throws Exception {
+        final ApplicationQueryListener listener = new ApplicationQueryListener(rlistener, appName+".query", resolver, sender, directMessagesExchangeName, converter, "globalReply", withDLQRetry, maxRetries, retryDelay, discardNotifier);
         listener.startListener();
         return listener;
     }
 
     @Bean
-    public ApplicationCommandListener applicationCommandListener(ReactiveMessageListener listener, HandlerResolver resolver, MessageConverter converter){
-        ApplicationCommandListener commandListener = new ApplicationCommandListener(listener, appName, resolver, directMessagesExchangeName, converter, withDLQRetry, maxRetries, retryDelay);
+    public ApplicationCommandListener applicationCommandListener(ReactiveMessageListener listener, HandlerResolver resolver, MessageConverter converter, DiscardNotifier discardNotifier){
+        ApplicationCommandListener commandListener = new ApplicationCommandListener(listener, appName, resolver, directMessagesExchangeName, converter, withDLQRetry, maxRetries, retryDelay, discardNotifier);
         commandListener.startListener();
         return commandListener;
     }