Add configuration number of threads in files processing.

Issue-ID: DCAEGEN2-2600
Signed-off-by: Tomasz Wrobel <tomasz.wrobel@nokia.com>
Change-Id: I04feb27698ef21196ab218de4bdd97be1fc85284

2 files changed, 102 insertions, 6 deletions

diff --git a/src/test/java/org/onap/dcaegen2/services/pmmapper/AppTest.java b/src/test/java/org/onap/dcaegen2/services/pmmapper/AppTest.java
index 0b8cdfc..db45029 100644
--- a/src/test/java/org/onap/dcaegen2/services/pmmapper/AppTest.java
+++ b/src/test/java/org/onap/dcaegen2/services/pmmapper/AppTest.java
@@ -1,6 +1,7 @@
 /*-
  * ============LICENSE_START=======================================================
  *  Copyright (C) 2019-2020 Nordix Foundation.
+ *  Copyright (C) 2021 Nokia.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -51,6 +52,7 @@ import org.onap.dcaegen2.services.pmmapper.exceptions.CBSServerError;
 import org.onap.dcaegen2.services.pmmapper.exceptions.EnvironmentConfigException;
 import org.onap.dcaegen2.services.pmmapper.exceptions.MapperConfigException;
 
+import org.onap.dcaegen2.services.pmmapper.config.FilesProcessingConfig;
 import org.onap.dcaegen2.services.pmmapper.utils.XMLValidator;
 import reactor.core.publisher.Flux;
 
@@ -76,6 +78,7 @@ import utils.EventUtils;
 @ExtendWith(MockitoExtension.class)
 class AppTest {
 
+    public static final int WANTED_NUMBER_OF_INVOCATIONS_1 = 1;
     static ClientAndServer mockServer;
     static MockServerClient client;
 
@@ -91,6 +94,7 @@ class AppTest {
 
     private App objUnderTest;
 
+    private final FilesProcessingConfig processingConfig = mock(FilesProcessingConfig.class);
 
     @BeforeAll
     static void setup() {
@@ -105,8 +109,10 @@ class AppTest {
     }
 
     @BeforeEach
-    void beforeEach() {
+    void beforeEach() throws EnvironmentConfigException {
         configHandler = mock(ConfigHandler.class);
+        when(this.processingConfig.getLimitRate()).thenReturn(1);
+        when(this.processingConfig.getThreadsCount()).thenReturn(1);
     }
 
     @Test
@@ -115,7 +121,7 @@ class AppTest {
         MapperConfig mockConfig = Mockito.spy(mapperConfig);
         when(mockConfig.getEnableHttp()).thenReturn(false);
         when(configHandler.getMapperConfig()).thenReturn(mockConfig);
-        objUnderTest = new App(template, schema, 0, 0, configHandler);
+        objUnderTest = new App(template, schema, 0, 0, configHandler, processingConfig);
         objUnderTest.start();
         assertEquals(1, objUnderTest.getApplicationServer().getListenerInfo().size());
         assertEquals("https", objUnderTest.getApplicationServer().getListenerInfo().get(0).getProtcol());
@@ -127,7 +133,7 @@ class AppTest {
         MapperConfig mockConfig = Mockito.spy(mapperConfig);
         when(mockConfig.getEnableHttp()).thenReturn(true);
         when(configHandler.getMapperConfig()).thenReturn(mockConfig);
-        objUnderTest = new App(template, schema, 0, 0, configHandler);
+        objUnderTest = new App(template, schema, 0, 0, configHandler, processingConfig);
         objUnderTest.start();
         assertEquals(2, objUnderTest.getApplicationServer().getListenerInfo().size());
         assertEquals("http", objUnderTest.getApplicationServer().getListenerInfo().get(0).getProtcol());
@@ -137,7 +143,7 @@ class AppTest {
     @Test
     void testConfigFailure() throws EnvironmentConfigException, CBSServerError, MapperConfigException {
         when(configHandler.getMapperConfig()).thenThrow(MapperConfigException.class);
-        assertThrows(IllegalStateException.class, () -> new App(template, schema, 0, 0, configHandler));
+        assertThrows(IllegalStateException.class, () -> new App(template, schema, 0, 0, configHandler, processingConfig));
 
     }
 
@@ -146,7 +152,7 @@ class AppTest {
         MapperConfig mockConfig = Mockito.spy(mapperConfig);
         when(mockConfig.getKeyStorePath()).thenReturn("not_a_file");
         when(configHandler.getMapperConfig()).thenReturn(mockConfig);
-        assertThrows(IllegalStateException.class, () -> new App(template, schema, 0, 0, configHandler));
+        assertThrows(IllegalStateException.class, () -> new App(template, schema, 0, 0, configHandler, processingConfig));
 
     }
 
@@ -319,4 +325,19 @@ class AppTest {
         Flux<List<Event>> mappingResult = App.map(new Mapper(mappingTemplate,mockMeasConverter), mockEvents, mockConfig);
         assertEquals(mappingResult, Flux.<List<Event>>empty());
     }
+
+    @Test
+    void filesProcessingConfiguration_IsReadInMainApp() throws Exception {
+        MapperConfig mockConfig = Mockito.spy(mapperConfig);
+        when(mockConfig.getEnableHttp()).thenReturn(true);
+        when(configHandler.getMapperConfig()).thenReturn(mockConfig);
+        objUnderTest = new App(template, schema, 0, 0, configHandler, processingConfig);
+        objUnderTest.start();
+
+        verify(processingConfig, times(WANTED_NUMBER_OF_INVOCATIONS_1)).getLimitRate();
+        verify(processingConfig, times(WANTED_NUMBER_OF_INVOCATIONS_1)).getThreadsCount();
+
+        objUnderTest.stop();
+    }
+
 }
diff --git a/src/test/java/org/onap/dcaegen2/services/pmmapper/config/FilesProcessingConfigTest.java b/src/test/java/org/onap/dcaegen2/services/pmmapper/config/FilesProcessingConfigTest.java
index fd21a39..b8d0a1f 100644
--- a/src/test/java/org/onap/dcaegen2/services/pmmapper/config/FilesProcessingConfigTest.java
+++ b/src/test/java/org/onap/dcaegen2/services/pmmapper/config/FilesProcessingConfigTest.java
@@ -32,7 +32,15 @@ public class FilesProcessingConfigTest {
 
     private static final String ENV_LIMIT_RATE = "PROCESSING_LIMIT_RATE";
 
-    private EnvironmentReader mockEnvironmentReader = mock(EnvironmentReader.class);
+    private static final String ENV_THREADS_MULTIPLIER = "THREADS_MULTIPLIER";
+    private static final String ENV_PROCESSING_THREADS_COUNT = "PROCESSING_THREADS_COUNT";
+    private static final String THREADS_4 = "4";
+    private static final String MULTIPLIER_3 = "3";
+
+    private static final int EXPECTED_4 = 4;
+    private static final int EXPECTED_12 = 12;
+
+    private final EnvironmentReader mockEnvironmentReader = mock(EnvironmentReader.class);
     private FilesProcessingConfig filesProcessingConfig;
 
     @Test
@@ -63,4 +71,71 @@ public class FilesProcessingConfigTest {
 
         assertEquals(1, limitRate);
     }
+
+    @Test
+    public void shouldReturnCorrectThreadsCount_whenVariableIsSet() throws EnvironmentConfigException {
+        when(mockEnvironmentReader.getVariable(ENV_PROCESSING_THREADS_COUNT)).thenReturn(THREADS_4);
+        when(mockEnvironmentReader.getVariable(ENV_THREADS_MULTIPLIER)).thenReturn(MULTIPLIER_3);
+
+        filesProcessingConfig = new FilesProcessingConfig(mockEnvironmentReader);
+        int threadsCount = filesProcessingConfig.getThreadsCount();
+
+        assertEquals(EXPECTED_12, threadsCount);
+    }
+
+    @Test
+    public void shouldReturnCorrectThreadsCount_whenVariableMultiplierIsNotSet() throws EnvironmentConfigException {
+
+        when(mockEnvironmentReader.getVariable(ENV_PROCESSING_THREADS_COUNT)).thenReturn(THREADS_4);
+
+        filesProcessingConfig = new FilesProcessingConfig(mockEnvironmentReader);
+        int threadsCount = filesProcessingConfig.getThreadsCount();
+
+        assertEquals(EXPECTED_4, threadsCount);
+    }
+
+    @Test
+    public void shouldReturnCorrectThreadsCount_whenVariableThreadsIsNotSet() throws EnvironmentConfigException {
+        when(mockEnvironmentReader.getVariable(ENV_THREADS_MULTIPLIER)).thenReturn(MULTIPLIER_3);
+
+        filesProcessingConfig = new FilesProcessingConfig(mockEnvironmentReader);
+        int threadsCount = filesProcessingConfig.getThreadsCount();
+        int expected = Runtime.getRuntime().availableProcessors() * Integer.parseInt(MULTIPLIER_3);
+
+        assertEquals(expected, threadsCount);
+    }
+
+    @Test
+    public void shouldReturnCorrectThreadsCount_whenVariableThreadsAndMultiplierIsNotSet()
+        throws EnvironmentConfigException {
+        filesProcessingConfig = new FilesProcessingConfig(mockEnvironmentReader);
+        int threadsCount = filesProcessingConfig.getThreadsCount();
+        int expected = Runtime.getRuntime().availableProcessors();
+
+        assertEquals(expected, threadsCount);
+    }
+
+    @Test
+    public void shouldThrowEnvironmentConfigException_whenProcessingThreadsVariableHasWrongValue() {
+        when(mockEnvironmentReader.getVariable(ENV_PROCESSING_THREADS_COUNT)).thenReturn("not-an-int");
+        filesProcessingConfig = new FilesProcessingConfig(mockEnvironmentReader);
+        String expectedMessage = "PROCESSING_THREADS_COUNT environment variable has incorrect value.\n";
+        String causeMessage = "For input string: \"not-an-int\"";
+
+        Throwable exception = assertThrows(EnvironmentConfigException.class, () -> filesProcessingConfig.getThreadsCount());
+        assertEquals(expectedMessage, exception.getMessage());
+        assertEquals(causeMessage, exception.getCause().getMessage());
+    }
+
+    @Test
+    public void shouldThrowEnvironmentConfigException_whenThreadsMultiplierVariableHasWrongValue() {
+        when(mockEnvironmentReader.getVariable(ENV_THREADS_MULTIPLIER)).thenReturn("not-an-int");
+        filesProcessingConfig = new FilesProcessingConfig(mockEnvironmentReader);
+        String expectedMessage = "THREADS_MULTIPLIER environment variable has incorrect value.\n";
+        String causeMessage = "For input string: \"not-an-int\"";
+
+        Throwable exception = assertThrows(EnvironmentConfigException.class, () -> filesProcessingConfig.getThreadsCount());
+        assertEquals(expectedMessage, exception.getMessage());
+        assertEquals(causeMessage, exception.getCause().getMessage());
+    }
 }