merge from master repo (#7)

Author: vasile-baluta

.travis.yml

@@ -13,7 +13,8 @@ before_install:
   - chmod +x pom.xml
 
 script:
-  - mvn -DsomeModule.test.excludes="**/FlowTest.java, **/FlowTest2.java, **/TrafficGeneratedTest.java" test
+  - mvn -DsomeModule.test.excludes="**/FlowTest.java, **/FlowTest2.java, **/TrafficGeneratedTest.java, **/FlowTestExternalComposition.java" test
   - mvn -DsomeModule.test.includes="**/FlowTest.java" test
   - mvn -DsomeModule.test.includes="**/FlowTest2.java" test
-  - mvn -DsomeModule.test.includes="**/TrafficGeneratedTest.java" test
+  - mvn -DsomeModule.test.includes="**/TrafficGeneratedTest.java" test
+  - mvn -DsomeModule.test.includes="**/FlowTestExternalComposition.java" test

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

@@ -36,17 +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;
-    }
-
     public void eventReceived(byte[] message) {
         log.info("Thread id " + Thread.currentThread().getId() + " spawned");
         String actualMessage = new String(message);
@@ -69,14 +58,4 @@ public void onMessage(Message message, Channel channel) throws Exception {
         int breakHere = 1;
     }
 
-    @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/RmqHandler.java

@@ -171,28 +171,16 @@ 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 bindToWaitlistQueueForRecentEvents(ConnectionFactory factory, EventHandler eventHandler) {
+    SimpleMessageListenerContainer bindToQueueForRecentEvents(ConnectionFactory factory, EventHandler eventHandler) {
+        String queueName = getQueueName();
         MessageListenerAdapter listenerAdapter = new EIMessageListenerAdapter(eventHandler);
-        waitlistContainer = new SimpleMessageListenerContainer();
-        waitlistContainer.setConnectionFactory(factory);
-        waitlistContainer.setQueueNames(getWaitlistQueueName());
-        waitlistContainer.setMessageListener(listenerAdapter);
-        waitlistContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);
-        return waitlistContainer;
+        container = new SimpleMessageListenerContainer();
+        container.setConnectionFactory(factory);
+        container.setQueueNames(queueName);
+        container.setMessageListener(listenerAdapter);
+        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
+        return container;
     }
 
     public String getQueueName() {
@@ -206,7 +194,7 @@ public String getWaitlistQueueName() {
     }
 
     @Bean
-    public RabbitTemplate waitListRabbitMqTemplate() {
+    public RabbitTemplate rabbitMqTemplate() {
         if (rabbitTemplate == null) {
             if (factory != null) {
                 rabbitTemplate = new RabbitTemplate(factory);
@@ -229,8 +217,7 @@ public void confirm(CorrelationData correlationData, boolean ack, String cause)
 
     public void publishObjectToWaitlistQueue(String message) {
         log.info("publishing message to message bus...");
-        //rabbitMqTemplate().convertAndSend(message);
-        waitListRabbitMqTemplate().convertAndSend(message);
+        rabbitMqTemplate().convertAndSend(message);
     }
 
     public void close() {

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

@@ -39,7 +39,7 @@ public class WaitListWorker {
 
     static Logger log = (Logger) LoggerFactory.getLogger(WaitListWorker.class);
 
-    @Scheduled(initialDelay = 10, fixedRate = 10)
+    @Scheduled(initialDelayString = "${waitlist.initialDelayResend}", fixedRateString = "${waitlist.fixedRateResend}")
     public void run() {
         RulesObject rulesObject = null;
         ArrayList<String> documents = waitListStorageHandler.getWaitList();

src/main/resources/application.properties

@@ -17,8 +17,6 @@ rabbitmq.queue.durable:true
 rabbitmq.routing.key:#
 rabbitmq.waitlist.queue.suffix:waitlist
 
-rabbitmq.waitlist.queue.suffix:waitlist
-
 mergeidmarker: %IdentifyRules%
 
 mongodb.host: localhost
@@ -28,6 +26,8 @@ aggregated.collection.name: aggregated_objects
 event_object_map.collection.name: event_object_map
 waitlist.collection.name: wait_list
 waitlist.collection.ttlValue: 600
+waitlist.initialDelayResend: 2000
+waitlist.fixedRateResend: 15000
 database.name: eiffel_intelligence_eiffelxxx
 subscription.collection.name: subscription
 
@@ -48,4 +48,4 @@ email.subject: Email Subscription Notification
 notification.failAttempt: 3
 notification.ttl.value: 600
 
-er.url: http://localhost:8080/search/ 
+er.url: http://localhost:8080/search/

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

@@ -1,47 +1,17 @@
 package com.ericsson.ei.flowtests;
 
-import com.ericsson.ei.handlers.ObjectHandler;
-import com.ericsson.ei.mongodbhandler.MongoDBHandler;
-import com.ericsson.ei.rmqhandler.RmqHandler;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.mongodb.MongoClient;
-import com.mongodb.client.MongoCollection;
-import com.mongodb.client.MongoDatabase;
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.ConnectionFactory;
-import de.flapdoodle.embed.mongo.distribution.Version;
-import de.flapdoodle.embed.mongo.tests.MongodForTestsFactory;
-import org.apache.commons.io.FileUtils;
-import org.apache.qpid.server.Broker;
-import org.apache.qpid.server.BrokerOptions;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-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.core.RabbitAdmin;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
+
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.TestPropertySource;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
-import javax.annotation.PostConstruct;
-import java.io.File;
 import java.util.ArrayList;
-import java.util.concurrent.TimeUnit;
 
-import static org.junit.Assert.assertEquals;
 
 @RunWith(SpringJUnit4ClassRunner.class)
-@TestPropertySource(properties = "myConf.myProp=valueInTest")
 @SpringBootTest
 public class FlowTest extends FlowTestBase {
 

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

@@ -1,42 +1,17 @@
 package com.ericsson.ei.flowtests;
 
-import com.ericsson.ei.handlers.ObjectHandler;
-import com.ericsson.ei.mongodbhandler.MongoDBHandler;
-import com.ericsson.ei.rmqhandler.RmqHandler;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.mongodb.MongoClient;
-import com.mongodb.client.MongoCollection;
-import com.mongodb.client.MongoDatabase;
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.ConnectionFactory;
-import de.flapdoodle.embed.mongo.distribution.Version;
-import de.flapdoodle.embed.mongo.tests.MongodForTestsFactory;
 import org.apache.commons.io.FileUtils;
-import org.apache.qpid.server.Broker;
-import org.apache.qpid.server.BrokerOptions;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-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.core.RabbitAdmin;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
+
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
-import javax.annotation.PostConstruct;
 import java.io.File;
 import java.util.ArrayList;
-import java.util.concurrent.TimeUnit;
-
 import static org.junit.Assert.assertEquals;
 
 @RunWith(SpringJUnit4ClassRunner.class)
@@ -64,20 +39,6 @@ protected ArrayList<String> getEventNamesToSend() {
         return eventNames;
     }
 
-    protected void waitForEventsToBeProcessed(int eventsCount) {
-        // wait for all events to be processed
-        int processedEvents = 0;
-        while (processedEvents < eventsCount) {
-            String countStr = System.getProperty("eiffel.intelligence.processedEventsCount");
-            String waitingCountStr = System.getProperty("eiffel.intelligence.waitListEventsCount");
-            if (waitingCountStr == null)
-                waitingCountStr = "0";
-            Properties props = admin.getQueueProperties(queue.getName());
-            int messageCount = Integer.parseInt(props.get("QUEUE_MESSAGE_COUNT").toString());
-            processedEvents = Integer.parseInt(countStr) - Integer.parseInt(waitingCountStr) - messageCount;
-        }
-    }
-
     protected void checkResult() {
         try {
             String document = objectHandler.findObjectById("6acc3c87-75e0-4b6d-88f5-b1a5d4e62b43");

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

@@ -5,7 +5,7 @@
 import java.io.File;
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.Properties;
+import java.util.concurrent.TimeUnit;
 
 import javax.annotation.PostConstruct;
 
@@ -23,13 +23,16 @@
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 
 import com.ericsson.ei.handlers.ObjectHandler;
 import com.ericsson.ei.mongodbhandler.MongoDBHandler;
 import com.ericsson.ei.rmqhandler.RmqHandler;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.mongodb.MongoClient;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.MongoDatabase;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
@@ -50,6 +53,9 @@ public class FlowTestBase {
     @Autowired
     private MongoDBHandler mongoDBHandler;
 
+    @Value("${database.name}") private String database;
+    @Value("${event_object_map.collection.name}") private String event_map;
+
     public static File qpidConfig = null;
     static AMQPBrokerManager amqpBrocker;
     protected static MongodForTestsFactory testsFactory;
@@ -106,7 +112,8 @@ public static void setUpMessageBus() throws Exception {
         System.setProperty("rabbitmq.port", "8672");
         System.setProperty("rabbitmq.user", "guest");
         System.setProperty("rabbitmq.password", "guest");
-
+        System.setProperty("waitlist.initialDelayResend", "10");
+        System.setProperty("waitlist.fixedRateResend", "10");
 
         String config = "src/test/resources/configs/qpidConfig.json";
         jsonFileContent = FileUtils.readFileToString(new File(jsonFilePath));
@@ -182,17 +189,25 @@ protected ArrayList<String> getEventNamesToSend() {
         return eventNames;
     }
 
+ // count documents that were processed
+    private long countProcessedEvents(String database, String collection){
+        MongoDatabase db = mongoClient.getDatabase(database);
+        MongoCollection table = db.getCollection(collection);
+        long countedDocuments = table.count();
+        return countedDocuments;
+    }
+
     protected void waitForEventsToBeProcessed(int eventsCount) {
-         int processedEvents = 0;
-         while (processedEvents < eventsCount) {
-             String countStr = System.getProperty("eiffel.intelligence.processedEventsCount");
-             String waitingCountStr = System.getProperty("eiffel.intelligence.waitListEventsCount");
-             if (waitingCountStr == null)
-                 waitingCountStr = "0";
-             Properties props = admin.getQueueProperties(queue.getName());
-             int messageCount = Integer.parseInt(props.get("QUEUE_MESSAGE_COUNT").toString());
-             processedEvents = Integer.parseInt(countStr) - Integer.parseInt(waitingCountStr) - messageCount;
-         }
+        // wait for all events to be processed
+        long processedEvents = 0;
+        while (processedEvents < eventsCount) {
+            processedEvents = countProcessedEvents(database, event_map);
+        }
+        try {
+            TimeUnit.MILLISECONDS.sleep(100);
+        } catch (Exception e) {
+            log.info(e.getMessage(),e);
+        }
     }
 
     protected void checkResult() {