Removing log dependency from ThreadingAndWaitlist test (#307)

* log dependency for the test is removed

* threads manipulation by setting names

Author: saif-ericsson

src/functionaltests/java/com/ericsson/ei/threadingAndWaitlistRepeat/ThreadingAndWaitlistRepeatSteps.java

@@ -3,12 +3,9 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotEquals;
 
-import java.io.File;
-import java.io.IOException;
-import java.io.PrintStream;
-import java.nio.file.Files;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -19,10 +16,12 @@
 import org.springframework.core.env.Environment;
 import org.springframework.test.context.TestPropertySource;
 
+import com.ericsson.ei.handlers.EventToObjectMapHandler;
+import com.ericsson.ei.rules.RulesObject;
 import com.ericsson.ei.utils.FunctionalTestBase;
-import com.ericsson.ei.waitlist.WaitListWorker;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 
-import cucumber.api.java.Before;
 import cucumber.api.java.en.Given;
 import cucumber.api.java.en.Then;
 
@@ -36,14 +35,16 @@
         "spring.data.mongodb.database: ThreadingAndWaitlistRepeatSteps",
         "missedNotificationDataBaseName: ThreadingAndWaitlistRepeatSteps-missedNotifications",
         "rabbitmq.exchange.name: ThreadingAndWaitlistRepeatSteps-exchange",
-        "rabbitmq.consumerName: ThreadingAndWaitlistRepeatStepsConsumer",
-        "logging.level.com.ericsson.ei.waitlist=DEBUG", "logging.level.com.ericsson.ei.handlers.EventHandler=DEBUG" })
+        "rabbitmq.consumerName: ThreadingAndWaitlistRepeatStepsConsumer", "logging.level.com.ericsson.ei.waitlist=OFF",
+        "logging.level.com.ericsson.ei.handlers.EventHandler=OFF" })
 
 @Ignore
 public class ThreadingAndWaitlistRepeatSteps extends FunctionalTestBase {
-
-    private File tempLogFile;
     private static final String EIFFEL_EVENTS_JSON_PATH = "src/functionaltests/resources/eiffel_events_for_thread_testing.json";
+    private static final String ID_RULE = "{" + "\"IdRule\": \"meta.id\"" + "}";
+
+    @Autowired
+    private Environment environment;
 
     @Value("${threads.corePoolSize}")
     private int corePoolSize;
@@ -54,18 +55,12 @@ public class ThreadingAndWaitlistRepeatSteps extends FunctionalTestBase {
     @Value("${waitlist.collection.ttlValue}")
     private int waitlistTtl;
 
-    @Autowired
-    WaitListWorker waitListWorker;
+    private RulesObject rulesObject;
 
-    @Autowired
-    Environment environment;
+    private JsonNode rulesJson;
 
-    @Before("@ThreadingAndWaitlistRepeatScenario")
-    public void beforeScenario() throws IOException {
-        tempLogFile = File.createTempFile("logfile", ".tmp");
-        tempLogFile.deleteOnExit();
-        System.setOut(new PrintStream(tempLogFile));
-    }
+    @Autowired
+    EventToObjectMapHandler eventToObjectMapHanler;
 
     @Given("^that eiffel events are sent$")
     public void that_eiffel_events_are_sent() throws Throwable {
@@ -86,45 +81,49 @@ public void no_event_is_aggregated() throws Throwable {
         assertEquals("aggregatedObjectExists was true, should be false, ", false, aggregatedObjectExists);
     }
 
-    @Then("^the waitlist will try to resend the events at given time interval$")
-    public void the_waitlist_will_try_to_resend_the_events_at_given_time_interval() throws Throwable {
-        TimeUnit.SECONDS.sleep(5);
-        List<String> resentEvents = new ArrayList<>();
-        List<String> lines = new ArrayList<>(Files.readAllLines(tempLogFile.toPath()));
-
-        int waitlistId = waitListWorker.hashCode();
-        String waitListIdPattern = "FROM WAITLIST: " + waitlistId;
-        Pattern pattern = Pattern.compile("\\[EIFFEL EVENT RESENT " + waitListIdPattern + "\\] id:([a-zA-Z\\d-]+)");
-        for (String line : lines) {
-            Matcher matcher = pattern.matcher(line);
-            if (matcher.find() && !matcher.group(1).equals("")) {
-                if (!resentEvents.contains(matcher.group(1))) {
-                    resentEvents.add(matcher.group(1));
-                }
-            }
+    @Then("^event-to-object-map is manipulated to include the sent events$")
+    public void event_to_object_map_is_manipulated_to_include_the_sent_events() throws Throwable {
+        JsonNode parsedJSON = eventManager.getJSONFromFile(EIFFEL_EVENTS_JSON_PATH);
+        ObjectMapper objectMapper = new ObjectMapper();
+        rulesJson = objectMapper.readTree(ID_RULE);
+        rulesObject = new RulesObject(rulesJson);
+
+        String dummyObjectID = "1234abcd-12ab-12ab-12ab-123456abcdef";
+        List<String> eventNames = getEventNamesToSend();
+        for (String eventName : eventNames) {
+            JsonNode eventJson = parsedJSON.get(eventName);
+            eventToObjectMapHanler.updateEventToObjectMapInMemoryDB(rulesObject, eventJson.toString(), dummyObjectID);
         }
-        assertEquals(getEventNamesToSend().size(), resentEvents.size());
+    }
+
+    @Then("^when waitlist has resent events they should have been deleted$")
+    public void when_waitlist_has_resent_events_they_should_have_been_deleted() throws Throwable {
+        long stopTime = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(3);
+        while (dbManager.waitListSize() > 0 && stopTime > System.currentTimeMillis()) {
+            TimeUnit.MILLISECONDS.sleep(100);
+        }
+        assertEquals("Waitlist resent events and due their presence in event-to-object-map, events are deleted", 0,
+                dbManager.waitListSize());
     }
 
     @Then("^correct amount of threads should be spawned$")
     public void correct_amount_of_threads_should_be_spawned() throws Throwable {
+        List<String> threadsSpawned = new ArrayList<>();
         String port = environment.getProperty("local.server.port");
-        String eventHandlerThreadIdPattern = String.format("Thread id (\\d+) spawned for EventHandler on port: %s",
-                port);
-        Pattern pattern = Pattern.compile(eventHandlerThreadIdPattern);
-        List<String> threadIds = new ArrayList<>();
-        List<String> lines = new ArrayList<>(Files.readAllLines(tempLogFile.toPath()));
-
-        for (String line : lines) {
-            Matcher matcher = pattern.matcher(line);
-            if (matcher.find() && !matcher.group(0).equals("")) {
-                if (!threadIds.contains(matcher.group(1))) {
-                    threadIds.add(matcher.group(1));
+        String eventHandlerThreadNamePattern = String.format("EventHandler-(\\d+)-%s", port);
+        Pattern pattern = Pattern.compile(eventHandlerThreadNamePattern);
+
+        Set<Thread> threadSet = Thread.getAllStackTraces().keySet();
+        Thread[] threadArray = threadSet.toArray(new Thread[threadSet.size()]);
+        for (Thread thread : threadArray) {
+            Matcher matcher = pattern.matcher(thread.getName());
+            if (matcher.find() && !matcher.group(1).equals("")) {
+                if (!threadsSpawned.contains(matcher.group(1))) {
+                    threadsSpawned.add(matcher.group(1));
                 }
             }
         }
-
-        assertEquals(getEventNamesToSend().size() - queueCapacity, threadIds.size());
+        assertEquals(getEventNamesToSend().size() - queueCapacity, threadsSpawned.size());
     }
 
     @Then("^after the time to live has ended, the waitlist should be empty$")

src/functionaltests/resources/features/threadingAndWaitlistRepeat.feature

@@ -6,6 +6,7 @@ Feature: Test Threading and Waitlist Repeat
   	Given that eiffel events are sent
 	Then waitlist should not be empty
 	And no event is aggregated
-	And the waitlist will try to resend the events at given time interval
+	And event-to-object-map is manipulated to include the sent events
+	And when waitlist has resent events they should have been deleted        
 	And correct amount of threads should be spawned
 	And after the time to live has ended, the waitlist should be empty

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

@@ -54,7 +54,6 @@ public RulesHandler getRulesHandler() {
     public void eventReceived(String event) {
         RulesObject eventRules = rulesHandler.getRulesForEvent(event);
         idRulesHandler.runIdRules(eventRules, event);
-        // downstreamIdRulesHandler.runIdRules(eventRules, event);
     }
 
     @Async
@@ -64,6 +63,7 @@ public void onMessage(Message message, Channel channel) throws Exception {
         JsonNode node = objectMapper.readTree(messageBody);
         String id = node.get("meta").get("id").toString();
         String port = environment.getProperty("local.server.port");
+        Thread.currentThread().setName(Thread.currentThread().getName() + "-" + port);
         LOGGER.debug("Thread id {} spawned for EventHandler on port: {}", Thread.currentThread().getId(), port);
         LOGGER.debug("Event {} received on port {}", id, port);
 

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

@@ -70,8 +70,6 @@ public void run() {
                     checkTargetAggregationsExistAndRepublishEvent(eventJson);
                 }
             } catch (Exception e) {
-                // LOGGER.error("Exception occured while trying to resend event: " + document,
-                // e);
                 LOGGER.error("Exception occured while trying to resend event: {}", document, e);
             }
         }

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

@@ -155,7 +155,8 @@ public void flowTest() {
      * are related. Events are sent to RabbitMQ queue. Deterministic traffic is
      * used.
      *
-     * @param eventNames list of events to be sent.
+     * @param eventNames
+     *            list of events to be sent.
      * @return list of ready to send events.
      */
     private List<String> getPreparedEventsToSend(List<String> eventNames) throws IOException {