debugging multiple subscription notification when repeat flag is off (#309)

saif-ericsson · web-flow · commit f6fca0657ea6 · 2019-08-20T16:45:29.000+02:00
* adding synchronized block to avoid race condition
diff --git a/src/functionaltests/java/com/ericsson/ei/notifications/trigger/SubscriptionNotificationSteps.java b/src/functionaltests/java/com/ericsson/ei/notifications/trigger/SubscriptionNotificationSteps.java
diff --git a/src/main/java/com/ericsson/ei/subscription/RunSubscription.java b/src/main/java/com/ericsson/ei/subscription/RunSubscription.java
@@ -21,12 +21,15 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 
 import com.ericsson.ei.jmespath.JmesPathInterface;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ArrayNode;
 
+import lombok.Getter;
+
 /**
  * This class represents the mechanism to fetch the rule conditions from the
  * Subscription Object and match it with the aggregatedObject to check if it is
@@ -45,6 +48,10 @@ public class RunSubscription {
 
     @Autowired
     private SubscriptionRepeatDbHandler subscriptionRepeatDbHandler;
+    
+    @Getter
+    @Value("${spring.data.mongodb.database}")
+    public String dataBaseName;
 
     /**
      * This method matches every condition specified in the subscription Object and
@@ -71,15 +78,15 @@ public boolean runSubscriptionOnObject(String aggregatedObject, Iterator<JsonNod
             String subscriptionRepeatFlag = subscriptionJson.get("repeat").asText();
 
             if (id == null) {
-                LOGGER.error(
-                    "ID has not been passed for given aggregated object. The subscription will be triggered again.");
+                LOGGER.debug(
+                        "ID has not been passed for given aggregated object. The subscription will be triggered again.");
             }
 
-            if (subscriptionRepeatFlag == "false" && id != null
-                    && subscriptionRepeatDbHandler.checkIfAggrObjIdExistInSubscriptionAggrIdsMatchedList(
-                            subscriptionName, requirementIndex, id)) {
-                LOGGER.info("Subscription has already matched with AggregatedObject Id: {}\n"
-                        + "SubscriptionName: {}\nand has Subscription Repeat flag set to: {}",
+            if (subscriptionRepeatFlag.equals("false") && id != null && subscriptionRepeatDbHandler
+                    .checkIfAggrObjIdExistInSubscriptionAggrIdsMatchedList(subscriptionName, requirementIndex, id)) {
+                LOGGER.debug(
+                        "Subscription has already matched with AggregatedObject Id: {}\n"
+                                + "SubscriptionName: {}\nand has Subscription Repeat flag set to: {}",
                         id, subscriptionName, subscriptionRepeatFlag);
                 break;
             }
@@ -103,19 +110,28 @@ public boolean runSubscriptionOnObject(String aggregatedObject, Iterator<JsonNod
                 boolean resultNotEmpty = !resultString.equals("");
                 boolean isFulfilled = resultNotEqualsToNull && resultNotEqualsToFalse && resultNotEmpty;
                 String fulfilledStatement = String.format("Condition was %sfulfilled.", isFulfilled ? "" : "not ");
-                LOGGER.debug("Condition: {}\nJMESPath evaluation result: {}\n{}",
-                        condition, result.toString(), fulfilledStatement);
+                LOGGER.debug("Condition: {}\nJMESPath evaluation result: {}\n{}", condition, result.toString(),
+                        fulfilledStatement);
                 if (resultNotEqualsToNull && resultNotEqualsToFalse && resultNotEmpty) {
                     count_condition_fulfillment++;
                 }
             }
 
             if (count_conditions != 0 && count_condition_fulfillment == count_conditions) {
                 conditionFulfilled = true;
-                if (subscriptionJson.get("repeat").toString() == "false" && id != null) {
-                    LOGGER.debug("Adding matched AggrObj id to SubscriptionRepeatFlagHandlerDb.");
-                    subscriptionRepeatDbHandler.addMatchedAggrObjToSubscriptionId(subscriptionName,
-                            requirementIndex, id);
+                if (subscriptionRepeatFlag.equals("false") && id != null) {
+                    // the keyword 'synchronized' ensures that this part of the code run
+                    // synchronously. Thus avoids race condition.
+                    synchronized (this) {
+                        if (!subscriptionRepeatDbHandler.checkIfAggrObjIdExistInSubscriptionAggrIdsMatchedList(
+                                subscriptionName, requirementIndex, id)) {
+                            LOGGER.debug("Adding matched aggregated object to database:" + dataBaseName);
+                            subscriptionRepeatDbHandler.addMatchedAggrObjToSubscriptionId(subscriptionName,
+                                    requirementIndex, id);
+                        } else {
+                            conditionFulfilled = false;
+                        }
+                    }
                 }
             }
 
diff --git a/src/test/java/com/ericsson/ei/flowtests/FlowSourceChangeObject.java b/src/test/java/com/ericsson/ei/flowtests/FlowSourceChangeObject.java
@@ -32,7 +32,11 @@
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootContextLoader;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.TestExecutionListeners;
 import org.springframework.test.context.TestPropertySource;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -42,7 +46,7 @@
 import com.ericsson.ei.erqueryservice.ERQueryService;
 import com.ericsson.ei.erqueryservice.SearchOption;
 import com.ericsson.ei.handlers.UpStreamEventsHandler;
-import com.ericsson.eiffelcommons.utils.ResponseEntity;
+import com.ericsson.ei.utils.TestContextInitializer;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -56,6 +60,7 @@
         "missedNotificationDataBaseName: FlowSourceChangeObject-missedNotifications",
         "rabbitmq.exchange.name: FlowSourceChangeObject-exchange",
         "rabbitmq.consumerName: FlowSourceChangeObjectConsumer" })
+@ContextConfiguration(classes = App.class, loader = SpringBootContextLoader.class, initializers = TestContextInitializer.class)
 public class FlowSourceChangeObject extends FlowTestBase {
     private static final String EVENTS_FILE_PATH = "src/test/resources/TestSourceChangeObjectEvents.json";
     private static final String AGGREGATED_OBJECT_FILE_PATH = "src/test/resources/aggregatedSourceChangeObject.json";
diff --git a/src/test/java/com/ericsson/ei/flowtests/FlowTest.java b/src/test/java/com/ericsson/ei/flowtests/FlowTest.java
@@ -65,6 +65,7 @@
         "missedNotificationDataBaseName: FlowTest-missedNotifications",
         "rabbitmq.exchange.name: FlowTest-exchange",
         "rabbitmq.consumerName: FlowTestConsumer" })
+
 public class FlowTest extends FlowTestBase {
 
     private static final String UPSTREAM_RESULT_FILE = "upStreamResultFile.json";
diff --git a/src/test/java/com/ericsson/ei/flowtests/FlowTest2.java b/src/test/java/com/ericsson/ei/flowtests/FlowTest2.java
@@ -45,6 +45,7 @@
         "missedNotificationDataBaseName: FlowTest2-missedNotifications",
         "rabbitmq.exchange.name: FlowTest2-exchange",
         "rabbitmq.consumerName: FlowTest2Consumer" })
+
 public class FlowTest2 extends FlowTestBase {
     private static final String EVENTS_FILE_PATH = "src/test/resources/test_events.json";
     private static final String AGGREGATED_OBJECT_FILE_PATH_1 = "src/test/resources/AggregatedDocument.json";
diff --git a/src/test/java/com/ericsson/ei/flowtests/FlowTestBase.java b/src/test/java/com/ericsson/ei/flowtests/FlowTestBase.java
@@ -131,13 +131,12 @@ protected int extraEventsCount() {
     public void flowTest() throws Exception {
         try {
             String queueName = rmqHandler.getQueueName();
-            Channel channel = TestConfigs.getConn().createChannel();
+            Channel channel = TestConfigs.getConnection().createChannel();
             if (channel == null) {
                 channel = connectionFactory.createConnection().createChannel(true);
             }
 
             String exchangeName = rmqHandler.getExchangeName();
-
             List<String> eventNames = getEventNamesToSend();
             JsonNode parsedJSON = getJSONFromFile(getEventsFilePath());
             int eventsCount = eventNames.size() + extraEventsCount();
diff --git a/src/test/java/com/ericsson/ei/flowtests/TrafficGeneratedTest.java b/src/test/java/com/ericsson/ei/flowtests/TrafficGeneratedTest.java
@@ -42,7 +42,11 @@
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.test.context.SpringBootContextLoader;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.TestExecutionListeners;
 import org.springframework.test.context.TestPropertySource;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -55,7 +59,7 @@
 import com.ericsson.ei.handlers.RmqHandler;
 import com.ericsson.ei.handlers.UpStreamEventsHandler;
 import com.ericsson.ei.test.utils.TestConfigs;
-import com.ericsson.eiffelcommons.utils.ResponseEntity;
+import com.ericsson.ei.utils.TestContextInitializer;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -71,6 +75,7 @@
         "missedNotificationDataBaseName: TrafficGeneratedTest-missedNotifications",
         "rabbitmq.exchange.name: TrafficGeneratedTest-exchange",
         "rabbitmq.consumerName: TrafficGeneratedTestConsumer"  })
+@ContextConfiguration(classes = App.class, loader = SpringBootContextLoader.class, initializers = TestContextInitializer.class)
 public class TrafficGeneratedTest extends FlowTestBase {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(TrafficGeneratedTest.class);
@@ -122,7 +127,7 @@ public void flowTest() {
             String queueName = rmqHandler.getQueueName();
             String exchange = "ei-poc-4";
             TestConfigs.createExchange(exchange, queueName);
-            Channel channel = TestConfigs.getConn().createChannel();
+            Channel channel = TestConfigs.getConnection().createChannel();
 
             long timeBefore = System.currentTimeMillis();
 
diff --git a/src/test/java/com/ericsson/ei/test/utils/TestConfigs.java b/src/test/java/com/ericsson/ei/test/utils/TestConfigs.java
@@ -2,6 +2,7 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.util.concurrent.TimeoutException;
 
 import org.apache.tomcat.util.codec.binary.Base64;
 import org.apache.tomcat.util.codec.binary.StringUtils;
@@ -30,12 +31,12 @@ public class TestConfigs {
     private static MongodForTestsFactory testsFactory;
 
     @Getter
-    private static ConnectionFactory cf;
+    private static ConnectionFactory connectionFactory;
 
     final static Logger LOGGER = LoggerFactory.getLogger(TestConfigs.class);
 
     @Getter
-    private static Connection conn;
+    private static Connection connection;
 
     @Getter
     private static MongoClient mongoClient = null;
@@ -46,42 +47,24 @@ public static synchronized void init() throws Exception {
     }
 
     private static synchronized void setUpMessageBus() throws Exception {
-        LOGGER.debug("Debug:setting up message buss");
-
-        LOGGER.debug("before setting up message buss: amqpBroker: " + amqpBroker + ", conn: " + conn + ",cf:" + cf);
-        if (amqpBroker != null || conn != null || cf != null) {
+        LOGGER.debug("Before setting up message bus, amqp broker: " + amqpBroker + ", connection: " + connection + ",connection factory:"
+                + connectionFactory);
+        if (amqpBroker != null || connection != null || connectionFactory != null) {
             return;
         }
 
         int port = SocketUtils.findAvailableTcpPort();
-        System.setProperty("rabbitmq.port", "" + port);
-        System.setProperty("rabbitmq.user", "guest");
-        System.setProperty("rabbitmq.password", "guest");
-        System.setProperty("waitlist.initialDelayResend", "500");
-        System.setProperty("waitlist.fixedRateResend", "100");
+        setSystemProperties(port);
+        setupBroker(port);
 
-        LOGGER.debug("done setting up message buss properties");
-        LOGGER.info("setting up message buss");
-        String config = "src/test/resources/configs/qpidConfig.json";
-        File qpidConfig = new File(config);
-        amqpBroker = new AMQPBrokerManager(qpidConfig.getAbsolutePath(), port);
-        amqpBroker.startBroker();
-        cf = new ConnectionFactory();
-        cf.setUsername("guest");
-        cf.setPassword("guest");
-
-        cf.setPort(port);
-        cf.setHandshakeTimeout(600000);
-        cf.setConnectionTimeout(600000);
-        conn = cf.newConnection();
-        LOGGER.debug("after setting up message buss");
+        setupConnectionFactory(port);
+        LOGGER.debug("Setting up message bus done!");
     }
 
     public static MongoClient mongoClientInstance() throws Exception {
         if (mongoClient == null) {
             setUpEmbeddedMongo();
         }
-
         return mongoClient;
     }
 
@@ -102,12 +85,13 @@ private static synchronized void setUpEmbeddedMongo() throws IOException {
     }
 
     public static void createExchange(final String exchangeName, final String queueName) {
-        final CachingConnectionFactory ccf = new CachingConnectionFactory(cf);
+        final CachingConnectionFactory ccf = new CachingConnectionFactory(connectionFactory);
         LOGGER.info("Creating exchange: {} and queue: {}", exchangeName, queueName);
-        RabbitAdmin admin = new RabbitAdmin(ccf);
-        Queue queue = new Queue(queueName, false);
-        admin.declareQueue(queue);
+        final RabbitAdmin admin = new RabbitAdmin(ccf);
+        final Queue queue = new Queue(queueName, false);
         final TopicExchange exchange = new TopicExchange(exchangeName);
+
+        admin.declareQueue(queue);
         admin.declareExchange(exchange);
         admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with("#"));
         ccf.destroy();
@@ -122,4 +106,29 @@ protected void setRules() {
         System.setProperty("rules", " /rules/ArtifactRules-Eiffel-Agen-Version.json");
     }
 
+    protected static void setSystemProperties(int port) {
+        System.setProperty("rabbitmq.port", "" + port);
+        System.setProperty("rabbitmq.user", "guest");
+        System.setProperty("rabbitmq.password", "guest");
+        System.setProperty("waitlist.initialDelayResend", "500");
+        System.setProperty("waitlist.fixedRateResend", "100");
+        LOGGER.info("Message bus port:{}", port);
+    }
+
+    protected static void setupBroker(int port) throws Exception {
+        String config = "src/test/resources/configs/qpidConfig.json";
+        File qpidConfig = new File(config);
+        amqpBroker = new AMQPBrokerManager(qpidConfig.getAbsolutePath(), port);
+        amqpBroker.startBroker();
+    }
+
+    protected static void setupConnectionFactory(int port) throws IOException, TimeoutException {
+        connectionFactory = new ConnectionFactory();
+        connectionFactory.setUsername("guest");
+        connectionFactory.setPassword("guest");
+        connectionFactory.setPort(port);
+        connectionFactory.setHandshakeTimeout(600000);
+        connectionFactory.setConnectionTimeout(600000);
+        connection = connectionFactory.newConnection();
+    }
 }
diff --git a/src/test/java/com/ericsson/ei/waitlist/TestWaitListWorker.java b/src/test/java/com/ericsson/ei/waitlist/TestWaitListWorker.java
@@ -145,7 +145,7 @@ public void testRunIfEventExistsInEventObjectMap() {
     @Test
     public void testPublishAndReceiveEvent() {
         try {
-            Channel channel = TestConfigs.getConn().createChannel();
+            Channel channel = TestConfigs.getConnection().createChannel();
             String queueName = "er001-eiffelxxx.eiffelintelligence.messageConsumer.durable";
             String exchange = "ei-poc-4";
             createExchange(exchange, queueName);
@@ -168,7 +168,7 @@ public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProp
     }
 
     private void createExchange(final String exchangeName, final String queueName) {
-        final CachingConnectionFactory ccf = new CachingConnectionFactory(TestConfigs.getCf());
+        final CachingConnectionFactory ccf = new CachingConnectionFactory(TestConfigs.getConnectionFactory());
         RabbitAdmin admin = new RabbitAdmin(ccf);
         Queue queue = new Queue(queueName, false);
         admin.declareQueue(queue);
