remove old bindings of a queue from message bus (#471)

* Added code changes for removing unused bindings of a queue from message bus

Co-authored-by: @Navyatha-reddi

Author: durga-vasaadi

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

@@ -7,9 +7,12 @@
 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.Binding.DestinationType;
 import org.springframework.amqp.core.BindingBuilder;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.core.TopicExchange;
@@ -28,8 +31,13 @@
 import com.ericsson.ei.handlers.EventHandler;
 import com.ericsson.ei.handlers.RMQHandler;
 import com.ericsson.ei.handlers.RMQProperties;
+import com.ericsson.ei.mongo.MongoCondition;
+import com.ericsson.ei.mongo.MongoDBHandler;
+import com.ericsson.ei.mongo.MongoQuery;
+import com.ericsson.ei.mongo.MongoStringQuery;
 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;
@@ -38,6 +46,7 @@
 @Ignore
 @TestPropertySource(properties = {
         "spring.data.mongodb.database: RabbitMQTestConnectionSteps",
+        "bindingkeys.collection.name: RabbitMQConfigurationTestSteps-bindingKeys",
         "failed.notifications.collection.name: RabbitMQTestConnectionSteps-failedNotifications",
         "rabbitmq.exchange.name: RabbitMQTestConnectionSteps-exchange",
         "rabbitmq.queue.suffix: RabbitMQTestConnectionSteps" })
@@ -49,6 +58,9 @@ public class RabbitMQTestConnectionSteps extends FunctionalTestBase {
     private static final Logger LOGGER = LoggerFactory.getLogger(RabbitMQTestConnectionSteps.class);
     private static final String EIFFEL_EVENTS = "src/functionaltests/resources/eiffel_events_for_thread_testing.json";
 
+    private static final String BINDING_KEY_1 = "binding-key-1";
+    private static final String BINDING_KEY_2 = "binding-key-2";
+
     private AMQPBrokerManager amqpBroker;
 
     @Autowired
@@ -58,6 +70,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();
@@ -116,6 +137,19 @@ public void can_send_events_which_are_put_in_the_waitlist() throws Exception {
         assertEquals(4, waitListSize);
     }
 
+    @When("^add the binding documents to mongoDB$")
+    public void add_the_binding_documents_to_mongoDB() {
+        BasicDBObject dbBinding = insertBinding();
+        assertEquals(5,dbBinding.size());
+    }
+
+    @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.
@@ -150,4 +184,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);
+            final MongoCondition condition = MongoCondition.bindingKeyCondition(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/integrationtests/resources/application.properties

@@ -26,6 +26,8 @@ rabbitmq.queue.durable: true
 rabbitmq.binding.key: #
 rabbitmq.waitlist.queue.suffix: waitList
 
+bindingkeys.collection.name: binding_keys
+
 encrypted.mongodb.password: ENC(en2h9ZQlZEuWyyfiTVrOPcQbRTwcXxOJ)
 spring.data.mongodb.uri: mongodb://admin:${encrypted.mongodb.password}@localhost:27016
 spring.data.mongodb.database: eiffel_intelligence

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

@@ -17,18 +17,23 @@
 package com.ericsson.ei.handlers;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
 import org.apache.commons.lang3.StringUtils;
+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;
@@ -41,6 +46,11 @@
 
 import com.ericsson.ei.listeners.EIMessageListenerAdapter;
 import com.ericsson.ei.listeners.RMQConnectionListener;
+import com.ericsson.ei.mongo.MongoCondition;
+import com.ericsson.ei.mongo.MongoConstants;
+import com.ericsson.ei.mongo.MongoDBHandler;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.mongodb.BasicDBObject;
 
 import lombok.Getter;
 import lombok.Setter;
@@ -68,6 +78,24 @@ public class RMQHandler {
     @Autowired
     private RMQProperties rmqProperties;
 
+    @Getter
+    @Setter
+    @Value("${spring.data.mongodb.database}")
+    private String dataBaseName;
+
+    @Getter
+    @Setter
+    @Value("${bindingkeys.collection.name}")
+    private String collectionName;
+
+    @Setter
+    @Autowired
+    private MongoDBHandler mongoDBHandler;
+
+    @Getter
+    @JsonIgnore
+    private AmqpAdmin amqpAdmin;
+
     @Bean
     public ConnectionFactory connectionFactory() {
         cachingConnectionFactory = new CachingConnectionFactory(rmqProperties.getHost(), rmqProperties.getPort());
@@ -169,11 +197,12 @@ protected Binding binding() {
 
     @Bean
     public List<Binding> bindings() {
-        String[] bingingKeysArray = splitBindingKeys(rmqProperties.getBindingKeys());
+        String[] bindingKeysArray = splitBindingKeys(rmqProperties.getBindingKeys());
         List<Binding> bindingList = new ArrayList<Binding>();
-        for (String bindingKey : bingingKeysArray) {
+        for (String bindingKey : bindingKeysArray) {
             bindingList.add(BindingBuilder.bind(externalQueue()).to(exchange()).with(bindingKey));
         }
+        deleteBindings(bindingKeysArray,bindingList);
         return bindingList;
     }
 
@@ -185,4 +214,61 @@ private String[] splitBindingKeys(String bindingKeys) {
         String bindingKeysWithoutWhitespace = bindingKeys.replaceAll("\\s+", "");
         return bindingKeysWithoutWhitespace.split(",");
     }
+
+    /**
+     * 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");
+                MongoCondition condition = MongoCondition.bindingKeyCondition(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 binding keys used for rabbitMQ, in mongo db.
+        storeNewBindingKeys(existingBindingsData, AMQPBindingObjectList);
+    }
+
+	/**
+     * This method is used to store the binding keys used for rabbitMQ, in 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(MongoConstants.MB_DESTINATION,bindingKey.getDestination());
+                    document.put(MongoConstants.MB_DESTINATIONT_TYPE, bindingKey.getDestinationType().toString());
+                    document.put(MongoConstants.MB_EXCHANGE, bindingKey.getExchange());
+                    document.put(MongoConstants.MB_BINDING_KEYS, bindingKey.getRoutingKey());
+                    document.put(MongoConstants.MB_ARG, bindingKey.getArguments().toString());
+                    mongoDBHandler.insertDocument(dataBaseName, collectionName, document.toString());
+             }
+         }
+    }
 }

src/main/java/com/ericsson/ei/mongo/MongoCondition.java

@@ -39,6 +39,18 @@ public static MongoCondition idCondition(String documentId) {
         return condition(MongoConstants.ID, documentId);
     }
 
+    /**
+     * Creates a MongoCondition to find a document with a given bindingKey. Called with
+     * <code>my-bindingKey</code> the JSON will look like this:
+     * <code>{"bindingKeys":"my-bindingKey"}</code>
+     *
+     * @param bindingKey The value of the bindingKey
+     * @return A MongoCondition with bindingKey set
+     */
+    public static MongoCondition bindingKeyCondition(String bindingKey) {
+        return condition(MongoConstants.MB_BINDING_KEYS, bindingKey);
+    }
+
     /**
      * Creates a MongoCodition just as {@link #idCondition(String)} but with the
      * <code>jsonNode</code> value fetched correctly.

src/main/java/com/ericsson/ei/mongo/MongoConstants.java

@@ -5,4 +5,10 @@ public class MongoConstants {
     public static final String EVENT = "Event";
     public static final String TIME = "Time";
     public static final String NOT_LOCKED = "0";
+
+    public static final String MB_DESTINATION = "destination";
+    public static final String MB_DESTINATIONT_TYPE = "destinationType";
+    public static final String MB_EXCHANGE = "exchange";
+    public static final String MB_BINDING_KEYS = "bindingKeys";
+    public static final String MB_ARG = "arg";
 }

src/main/resources/application.properties

@@ -26,6 +26,8 @@ rabbitmq.queue.durable: true
 rabbitmq.binding.key: #
 rabbitmq.waitlist.queue.suffix: waitList
 
+bindingkeys.collection.name: binding_keys
+
 spring.data.mongodb.uri: mongodb://localhost:27017
 spring.data.mongodb.database: eiffel_intelligence
 

wiki/configuration.md

@@ -204,6 +204,13 @@ The rabbitmq.tls.version property specifies the security protocol and you can fi
 * rabbitmq.binding.key
 * rabbitmq.waitlist.queue.suffix
 
+## Storage of rabbitMQ binding keys in mongo database
+
+Eiffel Intelligence stores bindings of a queue in a database with the collection name
+configured with the property **bindingkeys.collection.name**. Upon starting, Eiffel Intelligence compares,
+the bindingkeys with the property value rabbitmq.binding.key and bindings
+from mongoDB, then removes unused bindings from RabbitMQ and adds new bindings to the
+mongoDB collection. The name of this particular collection is defined by the below property:
 
 ## Security
 
@@ -296,3 +303,4 @@ Eiffel-Intelligence will decrypt password properties with help of the provided s
 Execute Eiffel-Intelligence with "jasypt.encryptor.password=mySecretEncryptorPassword" flag, example:
 
         java -jar eiffel-intelligence-<version>.war --jasypt.encryptor.password=mySecretEncryptorPassword --spring.config.location=/path/to/application.properties
+