message bus acknowledgement

Author: Vali (Vasile Baluta)

src/main/java/com/ericsson/ei/config/SpringAsyncConfig.java

@@ -0,0 +1,38 @@
+package com.ericsson.ei.config;
+
+import java.util.concurrent.Executor;
+
+import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.AsyncConfigurer;
+import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+@Configuration
+@EnableAsync
+public class SpringAsyncConfig implements AsyncConfigurer{
+
+    @Value("${threads.corePoolSize}") private int corePoolSize;
+    @Value("${threads.queueCapacity}") private int queueCapacity;
+    @Value("${threads.maxPoolSize}") private int maxPoolSize;
+
+
+    @Override
+    public Executor getAsyncExecutor() {
+         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+            executor.setCorePoolSize(corePoolSize);
+            executor.setQueueCapacity(queueCapacity);
+            executor.setMaxPoolSize(maxPoolSize);
+            executor.setThreadNamePrefix("EventHandler-");
+            executor.initialize();
+            return executor;
+    }
+
+
+    @Override
+    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
+        // TODO Auto-generated method stub
+        return null;
+    }
+}

src/main/java/com/ericsson/ei/handlers/EventHandler.java

@@ -18,13 +18,10 @@
 
 import java.util.concurrent.Executor;
 
-@Component
-public class EventHandler implements ChannelAwareMessageListener {
-
-    @Value("${threads.corePoolSize}") private int corePoolSize;
-    @Value("${threads.queueCapacity}") private int queueCapacity;
-    @Value("${threads.maxPoolSize}") private int maxPoolSize;
+import javax.annotation.PostConstruct;
 
+@Component
+public class EventHandler {
 
     private static Logger log = LoggerFactory.getLogger(EventHandler.class);
 
@@ -39,18 +36,6 @@ public void eventReceived(String event) {
         idRulesHandler.runIdRules(eventRules, event);
     }
 
-    @Bean
-    public Executor asyncExecutor() {
-        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
-        executor.setCorePoolSize(corePoolSize);
-        executor.setQueueCapacity(queueCapacity);
-        executor.setMaxPoolSize(maxPoolSize);
-        executor.setThreadNamePrefix("EventHandler-");
-        executor.initialize();
-        return executor;
-    }
-
-    @Async
     public void eventReceived(byte[] message) {
         log.info("Thread id " + Thread.currentThread().getId() + " spawned");
         String actualMessage = new String(message);
@@ -64,13 +49,14 @@ public void eventReceived(byte[] message) {
         }
     }
 
-    @Override
+    @Async
     public void onMessage(Message message, Channel channel) throws Exception {
         byte[] messageBody = message.getBody();
+//        String messageStr = new String(messageBody);
         eventReceived(messageBody);
         long deliveryTag = message.getMessageProperties().getDeliveryTag();
+//        String queue = message.getMessageProperties().getConsumerQueue();
         channel.basicAck(deliveryTag, false);
         int breakHere = 1;
-
     }
 }

src/main/java/com/ericsson/ei/rmqhandler/EIMessageListenerAdapter.java

@@ -0,0 +1,27 @@
+package com.ericsson.ei.rmqhandler;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
+
+import com.ericsson.ei.handlers.EventHandler;
+import com.rabbitmq.client.Channel;
+
+public class EIMessageListenerAdapter extends MessageListenerAdapter {
+
+    public EIMessageListenerAdapter(Object delegate) {
+        super(delegate);
+    }
+
+    @Override
+    public void onMessage(Message message, Channel channel) throws Exception {
+        Object delegate = getDelegate();
+        if (delegate != this) {
+            if (delegate instanceof EventHandler) {
+                if (channel != null) {
+                    ((EventHandler) delegate).onMessage(message, channel);
+                    return;
+                }
+            }
+        }
+    }
+}

src/main/java/com/ericsson/ei/rmqhandler/RmqHandler.java

@@ -46,6 +46,8 @@ public class RmqHandler {
     private String domainId;
     @Value("${rabbitmq.componentName}")
     private String componentName;
+    @Value("${rabbitmq.waitlist.queue.suffix}")
+    private String waitlistSufix;
     @Value("${rabbitmq.routing.key}")
     private String routingKey;
     @Value("${rabbitmq.consumerName}")
@@ -54,6 +56,7 @@ public class RmqHandler {
     private RabbitTemplate rabbitTemplate;
     private CachingConnectionFactory factory;
     private SimpleMessageListenerContainer container;
+    private SimpleMessageListenerContainer waitlistContainer;
     static Logger log = (Logger) LoggerFactory.getLogger(RmqHandler.class);
 
     public Boolean getQueueDurable() {
@@ -169,26 +172,42 @@ Binding binding(Queue queue, TopicExchange exchange) {
         return BindingBuilder.bind(queue).to(exchange).with(routingKey);
     }
 
+//    @Bean
+//    SimpleMessageListenerContainer bindToQueueForRecentEvents(ConnectionFactory factory, EventHandler eventHandler) {
+//        String queueName = getQueueName();
+////        MessageListenerAdapter listenerAdapter = new MessageListenerAdapter(eventHandler, "eventReceived");
+//        MessageListenerAdapter listenerAdapter = new EIMessageListenerAdapter(eventHandler);
+//        container = new SimpleMessageListenerContainer();
+//        container.setConnectionFactory(factory);
+//        container.setQueueNames(queueName);
+//        container.setMessageListener(listenerAdapter);
+////        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
+//        return container;
+//    }
+
     @Bean
-    SimpleMessageListenerContainer bindToQueueForRecentEvents(ConnectionFactory factory, EventHandler eventHandler) {
-        String queueName = getQueueName();
-//        MessageListenerAdapter listenerAdapter = new MessageListenerAdapter(eventHandler, "eventReceived");
-        MessageListenerAdapter listenerAdapter = new MessageListenerAdapter(eventHandler);
-        container = new SimpleMessageListenerContainer();
-        container.setConnectionFactory(factory);
-        container.setQueueNames(queueName);
-        container.setMessageListener(listenerAdapter);
-        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
-        return container;
+    SimpleMessageListenerContainer bindToWaitlistQueueForRecentEvents(ConnectionFactory factory, EventHandler eventHandler) {
+        MessageListenerAdapter listenerAdapter = new EIMessageListenerAdapter(eventHandler);
+        waitlistContainer = new SimpleMessageListenerContainer();
+        waitlistContainer.setConnectionFactory(factory);
+        waitlistContainer.setQueueNames(getWaitlistQueueName());
+        waitlistContainer.setMessageListener(listenerAdapter);
+        waitlistContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);
+        return waitlistContainer;
     }
 
     public String getQueueName() {
         String durableName = queueDurable ? "durable" : "transient";
         return domainId + "." + componentName + "." + consumerName + "." + durableName;
     }
 
+    public String getWaitlistQueueName() {
+        String durableName = queueDurable ? "durable" : "transient";
+        return domainId + "." + componentName + "." + consumerName + "." + durableName + "." + waitlistSufix;
+    }
+
     @Bean
-    public RabbitTemplate rabbitMqTemplate() {
+    public RabbitTemplate waitListRabbitMqTemplate() {
         if (rabbitTemplate == null) {
             if (factory != null) {
                 rabbitTemplate = new RabbitTemplate(factory);
@@ -209,22 +228,14 @@ public void confirm(CorrelationData correlationData, boolean ack, String cause)
         return rabbitTemplate;
     }
 
-    public void publishObjectToMessageBus(String message) {
+    public void publishObjectToWaitlistQueue(String message) {
         log.info("publishing message to message bus...");
-        rabbitMqTemplate().convertAndSend(message);
-//        Connection conn = factory.createConnection();
-//        Channel channel = conn.createChannel(true);
-//        String queueName = getQueueName();
-//        String exchange = exchangeName;
-//        try {
-//            channel.basicPublish(exchange, queueName, null, message.getBytes());
-//        } catch (Exception e) {
-//            log.info(e.getMessage(),e);
-//        }
+        waitListRabbitMqTemplate().convertAndSend(message);
     }
 
     public void close() {
         try {
+            waitlistContainer.destroy();
             container.destroy();
             factory.destroy();
         } catch (Exception e) {

src/main/java/com/ericsson/ei/waitlist/WaitListWorker.java

@@ -53,7 +53,7 @@ public void run() {
                 for (final JsonNode idJsonObj : ids) {
                     ArrayList<String> objects = matchIdRulesHandler.fetchObjectsById(rulesObject, idJsonObj.textValue());
                     if (objects.size() > 0) {
-                        rmqHandler.publishObjectToMessageBus(event);
+                        rmqHandler.publishObjectToWaitlistQueue(event);
                         waitListStorageHandler.dropDocumentFromWaitList(document);
                     }
                 }

src/main/resources/application.properties

@@ -16,6 +16,8 @@ rabbitmq.consumerName: messageConsumer
 rabbitmq.queue.durable:true
 rabbitmq.routing.key:#
 
+rabbitmq.waitlist.queue.suffix:waitlist
+
 mergeidmarker: %IdentifyRules%
 
 mongodb.host: localhost

src/test/java/com/ericsson/ei/flowtests/FlowTest.java

@@ -150,16 +150,15 @@ public void flowTest() {
         try {
             String queueName = rmqHandler.getQueueName();
             Channel channel = conn.createChannel();
-            String exchange = "ei-poc-4";
-            createExchange(exchange, queueName);
-
+            String exchangeName = "ei-poc-4";
+            createExchange();
 
             ArrayList<String> eventNames = getEventNamesToSend();
             int eventsCount = eventNames.size();
             for(String eventName : eventNames) {
                 JsonNode eventJson = parsedJason.get(eventName);
                 String event = eventJson.toString();
-                channel.basicPublish(exchange, queueName,  null, event.getBytes());
+                channel.basicPublish(exchangeName, queueName,  null, event.getBytes());
             }
 
             // wait for all events to be processed
@@ -197,11 +196,16 @@ private ArrayList<String> getEventNamesToSend() {
          return eventNames;
     }
 
-    private void createExchange(final String exchangeName, final String queueName) {
+    private void createExchange() {
+        String queueName = rmqHandler.getQueueName();
+        String waitlistQueueName = rmqHandler.getWaitlistQueueName();
+        String exchangeName = "ei-poc-4";
         final CachingConnectionFactory ccf = new CachingConnectionFactory(cf);
         admin = new RabbitAdmin(ccf);
-        queue = new Queue(queueName, false);
+        queue = new Queue(queueName, true);
+        Queue waitlistQueue = new Queue(waitlistQueueName, true);
         admin.declareQueue(queue);
+        admin.declareQueue(waitlistQueue);
         final TopicExchange exchange = new TopicExchange(exchangeName);
         admin.declareExchange(exchange);
         admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with("#"));