add support for manual akcnowledgement when event has been processed

Author: Vali (Vasile Baluta)

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

@@ -2,14 +2,17 @@
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 import com.ericsson.ei.rules.RulesHandler;
 import com.ericsson.ei.rules.RulesObject;
+import com.rabbitmq.client.Channel;
 
 @Component
-public class EventHandler {
+public class EventHandler implements ChannelAwareMessageListener {
 
     private static Logger log = LoggerFactory.getLogger(EventHandler.class);
 
@@ -35,4 +38,14 @@ public void eventReceived(byte[] message) {
             System.setProperty("eiffel.intelligence.processedEventsCount", "" + count);
         }
     }
+
+    @Override
+    public void onMessage(Message message, Channel channel) throws Exception {
+        byte[] messageBody = message.getBody();
+        eventReceived(messageBody);
+        long deliveryTag = message.getMessageProperties().getDeliveryTag();
+        channel.basicAck(deliveryTag, false);
+        int breakHere = 1;
+
+    }
 }

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

@@ -4,13 +4,15 @@
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.Binding;
 import org.springframework.amqp.core.BindingBuilder;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.core.TopicExchange;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.Connection;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
@@ -170,11 +172,13 @@ Binding binding(Queue queue, TopicExchange exchange) {
     @Bean
     SimpleMessageListenerContainer bindToQueueForRecentEvents(ConnectionFactory factory, EventHandler eventHandler) {
         String queueName = getQueueName();
-        MessageListenerAdapter listenerAdapter = new MessageListenerAdapter(eventHandler, "eventReceived");
+//        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;
     }