Old bindingkey is removed if binding key is changed in configuration (#468)

* Added property in configuration to handle the bindings in RabbitMQ when new bindings are added and old bindings are removed.
* Added functionality to handle the above property 
* Added documentation for the above change.

Author: Navyatha-reddi

pom.xml

@@ -6,7 +6,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.github.ericsson</groupId>
     <artifactId>eiffel-intelligence</artifactId>
-    <version>2.2.2</version>
+    <version>2.2.3</version>
     <packaging>war</packaging>
 
     <parent>

src/functionaltests/java/com/ericsson/ei/rabbitmq/configuration/RabbitMQConfigurationTestSteps.java

@@ -1,14 +1,18 @@
 package com.ericsson.ei.rabbitmq.configuration;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 
 import java.io.File;
 import java.util.ArrayList;
 import java.util.List;
 
+import org.json.JSONObject;
 import org.junit.Ignore;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.Binding.DestinationType;
 import org.springframework.amqp.core.BindingBuilder;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.core.TopicExchange;
@@ -23,9 +27,11 @@
 import org.springframework.util.SocketUtils;
 
 import com.ericsson.ei.handlers.EventHandler;
+import com.ericsson.ei.handlers.MongoDBHandler;
 import com.ericsson.ei.handlers.RmqHandler;
 import com.ericsson.ei.utils.AMQPBrokerManager;
 import com.ericsson.ei.utils.FunctionalTestBase;
+import com.mongodb.BasicDBObject;
 
 import cucumber.api.java.en.Given;
 import cucumber.api.java.en.Then;

src/functionaltests/java/com/ericsson/ei/rabbitmq/connection/RabbitMQTestConnectionSteps.java

@@ -1,18 +1,22 @@
 package com.ericsson.ei.rabbitmq.connection;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 
 import java.io.File;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
+import org.json.JSONObject;
 import org.junit.Ignore;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+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.core.Binding.DestinationType;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
@@ -24,9 +28,11 @@
 import org.springframework.util.SocketUtils;
 
 import com.ericsson.ei.handlers.EventHandler;
+import com.ericsson.ei.handlers.MongoDBHandler;
 import com.ericsson.ei.handlers.RmqHandler;
 import com.ericsson.ei.utils.AMQPBrokerManager;
 import com.ericsson.ei.utils.FunctionalTestBase;
+import com.mongodb.BasicDBObject;
 
 import cucumber.api.java.en.Given;
 import cucumber.api.java.en.Then;
@@ -35,6 +41,7 @@
 @Ignore
 @TestPropertySource(properties = {
         "spring.data.mongodb.database: RabbitMQTestConnectionSteps",
+        "bindingkeys.collection.name: RabbitMQConfigurationTestSteps-bindingKeys",
         "missedNotificationDataBaseName: RabbitMQTestConnectionSteps-missedNotifications",
         "rabbitmq.exchange.name: RabbitMQTestConnectionSteps-exchange",
         "rabbitmq.consumerName: RabbitMQTestConnectionStepsConsumer" })
@@ -48,6 +55,9 @@ public class RabbitMQTestConnectionSteps extends FunctionalTestBase {
 
     private static final String DEFAULT_ROUTING_KEY = "#";
 
+    private static final String BINDING_KEY_1 = "binding-key-1";
+    private static final String BINDING_KEY_2 = "binding-key-2";
+
     private AMQPBrokerManager amqpBroker;
 
     @Autowired
@@ -57,6 +67,15 @@ public class RabbitMQTestConnectionSteps extends FunctionalTestBase {
     @Autowired
     EventHandler eventHandler;
 
+    @Value("${spring.data.mongodb.database}")
+    private String dataBaseName;
+
+    @Value("${bindingkeys.collection.name}")
+    private String collectionName;
+
+    @Autowired
+    private MongoDBHandler mongoDBHandler;
+
     @Given("^We are connected to message bus$")
     public void connect_to_message_bus() throws Exception {
         int port = SocketUtils.findAvailableTcpPort();
@@ -108,6 +127,19 @@ public void can_send_events_which_are_put_in_the_waitlist() throws Exception {
         assertEquals(1, waitListSize);
     }
 
+    @When("^add the binding documents to mongoDB$")
+    public void add_the_binding_documents_to_mongoDB() {
+        BasicDBObject dbBinding = insertBinding();
+        assertNotNull(dbBinding);
+    }
+
+    @Then("^compare the binding keys and remove the old binding keys from rabbitMQ and mongoDB$")
+    public void compare_the_binding_keys_and_remove_the_old_binding_keys_from_rabbitMQ_and_mongoDB() {
+        LOGGER.debug("comparing the binding keys to remove the old binding key");
+        ArrayList<String> removedBinding = compareAndRemoveBindings();
+        assertEquals(1, removedBinding.size());
+    }
+
     /**
      * This method collects all the event names of events we will send to the
      * message bus.
@@ -138,4 +170,33 @@ private RabbitAdmin createExchange(final RmqHandler rmqHandler) {
         return admin;
     }
 
+    private BasicDBObject insertBinding() {
+        BasicDBObject docInput = new BasicDBObject();
+        docInput.put("destination", "ei-domain.eiffel-intelligence.messageConsumer.durable");
+        docInput.put("destinationType", "QUEUE");
+        docInput.put("exchange", "ei-binding-keys");
+        docInput.put("bindingKeys", BINDING_KEY_1);
+        docInput.put("arg", null);
+        mongoDBHandler.insertDocument(dataBaseName, collectionName, docInput.toString());
+        return docInput;
+    }
+
+    private ArrayList<String> compareAndRemoveBindings() {
+        ArrayList<Binding> listBinding = new ArrayList<Binding>();
+        listBinding.add(new Binding("ei-domain.eiffel-intelligence.messageConsumer.durable", DestinationType.QUEUE,
+                "ei-binding-keys", BINDING_KEY_2, null));
+        List<String> allObjects = mongoDBHandler.getAllDocuments(dataBaseName, collectionName);
+        ArrayList<String> removedBinding = new ArrayList<String>();
+
+        JSONObject dbBinding = new JSONObject(allObjects.get(0));
+        String mongoDbBinding = dbBinding.getString("bindingKeys");
+        if (!(listBinding.contains(mongoDbBinding))) {
+            removedBinding.add(mongoDbBinding);
+            listBinding.remove(0);
+            String condition = "{\"bindingKeys\": /.*" + mongoDbBinding + "/}";
+            mongoDBHandler.dropDocument(dataBaseName, collectionName, condition);
+        }
+        return removedBinding;
+    }
+
 }

src/functionaltests/resources/features/rabbitMQTestConnection.feature

@@ -7,3 +7,9 @@ Feature: Test Rabbit MQ Connection
     When Message bus goes down
     And Message bus is restarted
     Then I can send events which are put in the waitlist
+
+  @RabbitMQConfigurationDeleteBindingKeysScenario
+  Scenario: Test that old binding keys are deleted from rabbitMQ and mongoDB
+    Given We are connected to message bus
+    When add the binding documents to mongoDB
+    Then compare the binding keys and remove the old binding keys from rabbitMQ and mongoDB

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

@@ -17,28 +17,34 @@
 package com.ericsson.ei.handlers;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
-
+import org.json.JSONObject;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.amqp.core.AcknowledgeMode;
+import org.springframework.amqp.core.AmqpAdmin;
 import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.Binding.DestinationType;
 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.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
 import org.springframework.amqp.rabbit.support.CorrelationData;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.stereotype.Component;
 
 import com.ericsson.ei.listener.EIMessageListenerAdapter;
 import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.mongodb.BasicDBObject;
 
 import lombok.Getter;
 import lombok.Setter;
@@ -108,9 +114,23 @@ public class RmqHandler {
     @Setter
     @Value("${rabbitmq.consumerName}")
     private String consumerName;
+    
+    @Getter
+    @Setter
+    @Value("${spring.data.mongodb.database}")
+    private String dataBaseName;
+
+    @Getter
+    @Setter
+    @Value("${bindingkeys.collection.name}")
+    private String collectionName;
 
     @Value("${threads.maxPoolSize}")
     private int maxThreads;
+    
+    @Setter
+    @Autowired
+    private MongoDBHandler mongoDBHandler;
 
     @Setter
     @JsonIgnore
@@ -122,6 +142,10 @@ public class RmqHandler {
     @JsonIgnore
     private SimpleMessageListenerContainer container;
 
+    @Getter
+    @JsonIgnore
+    private AmqpAdmin amqpAdmin;
+
     @Bean
     public ConnectionFactory connectionFactory() {
         cachingConnectionFactory = new CachingConnectionFactory(host, port);
@@ -171,15 +195,75 @@ Binding binding() {
     }
 
     @Bean
-    public List<Binding> bindings() {
-        final String[] bingingKeysArray = splitBindingKeys(bindingKeys);
+    public List<Binding> bindings(){
+        final String[] bindingKeysArray = splitBindingKeys(bindingKeys);
         final List<Binding> bindingList = new ArrayList<>();
-        for (final String bindingKey : bingingKeysArray) {
+        for (final String bindingKey : bindingKeysArray) {
             bindingList.add(BindingBuilder.bind(externalQueue()).to(exchange()).with(bindingKey));
         }
+        deleteBindings(bindingKeysArray,bindingList);
         return bindingList;
     }
 
+    /**
+     * This method is used to delete the bindings in rabbitMQ.
+     * By comparing the binding keys used in the properties and binding keys stored in mongoDB.
+     * newBindingKeysArray is the only binding keys array.
+     * AMQPBindingObjectList is entire list of bindings.
+     * Binding key which is not present in the current AMQPBindingObjectList gets deleted and removed from mongoDB.
+     * @return
+     */
+
+    private void deleteBindings(String[] newBindingKeysArray, List<Binding> AMQPBindingObjectList) {
+        // Creating BindingKeys Collection in mongoDB
+        ArrayList<String> allDocuments = mongoDBHandler.getAllDocuments(dataBaseName, collectionName);
+        ArrayList<String> existingBindingsData = new ArrayList<String>();
+        if (!allDocuments.isEmpty()) {
+            for (String bindings : allDocuments) {
+                JSONObject bindingObj = new JSONObject(bindings);
+                final String mongoDbBindingKey = bindingObj.getString("bindingKeys");
+                String condition = "{\"bindingKeys\": /.*" + mongoDbBindingKey + "/}";
+                if (!Arrays.asList(newBindingKeysArray).contains(mongoDbBindingKey)) {
+                    String destinationDB = bindingObj.getString("destination");
+                    String exchangeDB = bindingObj.getString("exchange");
+                    // Binding the old binding key and removing from queue
+                    Binding b = new Binding(destinationDB, DestinationType.QUEUE, exchangeDB, mongoDbBindingKey, null);
+                    amqpAdmin = new RabbitAdmin(connectionFactory());
+                    amqpAdmin.removeBinding(b);
+                    // Removing binding document from mongoDB
+                    mongoDBHandler.dropDocument(dataBaseName, collectionName, condition);
+                } else {
+                    // storing the existing key into an array.
+                    existingBindingsData.add(mongoDbBindingKey);
+                }
+            }
+        }
+        // To store the new binding key into the mongoDB.
+        storeNewBindingKeys(existingBindingsData, AMQPBindingObjectList);
+    }
+
+    /**
+     * This method is used to store the bindings of new binding key into mongoDB.
+     * @return
+     */
+
+    private void storeNewBindingKeys(ArrayList<String> existingBindingsData, List<Binding> AMQPBindingObjectList){
+    // comparing with the stored key and adding the new binding key into the mongoDB.
+       for(final Binding bindingKey:AMQPBindingObjectList){
+            if(existingBindingsData.contains(bindingKey.getRoutingKey())){
+                LOGGER.info("Binding already present in mongoDB");
+            }else{
+                    BasicDBObject document = new BasicDBObject();
+                    document.put("destination",bindingKey.getDestination());
+                    document.put("destinationType", bindingKey.getDestinationType().toString());
+                    document.put("exchange", bindingKey.getExchange());
+                    document.put("bindingKeys", bindingKey.getRoutingKey());
+                    document.put("arg", bindingKey.getArguments().toString());
+                    mongoDBHandler.insertDocument(dataBaseName, collectionName, document.toString());
+             }
+         }
+    }
+
     @Bean
     public SimpleMessageListenerContainer bindToQueueForRecentEvents(
             final ConnectionFactory springConnectionFactory,

src/main/resources/application.properties

@@ -81,6 +81,8 @@ subscription.collection.name: subscription
 # name of the collection where we store pair of subscription:aggregated 
 # object where notification has been done and should not be repeated
 subscription.collection.repeatFlagHandlerName: subscription_repeat_handler
+# name of the collection where binding keys are stored
+bindingkeys.collection.name: binding_keys
 
 # Need to be set to True to use Test Rules page in EI-Frontend.
 # This is when you want to define new aggregation rules and test it