diff options
Diffstat (limited to 'appc-dispatcher/appc-dispatcher-common/execution-queue-management-lib')
2 files changed, 81 insertions, 19 deletions
diff --git a/appc-dispatcher/appc-dispatcher-common/execution-queue-management-lib/src/main/java/org/onap/appc/executionqueue/impl/QueueManager.java b/appc-dispatcher/appc-dispatcher-common/execution-queue-management-lib/src/main/java/org/onap/appc/executionqueue/impl/QueueManager.java index 8bad66bea..579f5d1b1 100644 --- a/appc-dispatcher/appc-dispatcher-common/execution-queue-management-lib/src/main/java/org/onap/appc/executionqueue/impl/QueueManager.java +++ b/appc-dispatcher/appc-dispatcher-common/execution-queue-management-lib/src/main/java/org/onap/appc/executionqueue/impl/QueueManager.java @@ -5,6 +5,8 @@ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved. * ================================================================================ * Copyright (C) 2017 Amdocs + * ================================================================================ + * Modifications Copyright (C) 2019 Ericsson * ============================================================================= * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -23,27 +25,24 @@ package org.onap.appc.executionqueue.impl; -import com.att.eelf.configuration.EELFLogger; -import com.att.eelf.configuration.EELFManager; -import org.onap.appc.executionqueue.MessageExpirationListener; -import org.onap.appc.executionqueue.helper.Util; -import org.onap.appc.executionqueue.impl.object.QueueMessage; - import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import org.onap.appc.executionqueue.helper.Util; +import org.onap.appc.executionqueue.impl.object.QueueMessage; +import com.att.eelf.configuration.EELFLogger; +import com.att.eelf.configuration.EELFManager; public class QueueManager { private final EELFLogger logger = EELFManager.getInstance().getLogger(QueueManager.class); private ExecutorService messageExecutor; - private LinkedBlockingQueue<QueueMessage> queue; - private int max_thread_size; - private int max_queue_size; + private int maxThreadSize; + private int maxQueueSize; private Util executionQueueUtil; public QueueManager() { @@ -54,14 +53,14 @@ public class QueueManager { * Initialization method used by blueprint */ public void init() { - max_thread_size = executionQueueUtil.getThreadPoolSize(); - max_queue_size = executionQueueUtil.getExecutionQueueSize(); + maxThreadSize = executionQueueUtil.getThreadPoolSize(); + maxQueueSize = executionQueueUtil.getExecutionQueueSize(); messageExecutor = new ThreadPoolExecutor( - max_thread_size, - max_thread_size, + maxThreadSize, + maxThreadSize, 0L, TimeUnit.MILLISECONDS, - new LinkedBlockingQueue(max_queue_size), + new LinkedBlockingQueue(maxQueueSize), executionQueueUtil.getThreadFactory(true, "appc-dispatcher"), new ThreadPoolExecutor.AbortPolicy()); } @@ -108,9 +107,4 @@ public class QueueManager { return isEnqueued; } - - private boolean messageExpired(QueueMessage queueMessage) { - return queueMessage.getExpirationTime() != null && - queueMessage.getExpirationTime().getTime() < System.currentTimeMillis(); - } } diff --git a/appc-dispatcher/appc-dispatcher-common/execution-queue-management-lib/src/test/java/org/onap/appc/executionqueue/impl/QueueManagerTest.java b/appc-dispatcher/appc-dispatcher-common/execution-queue-management-lib/src/test/java/org/onap/appc/executionqueue/impl/QueueManagerTest.java new file mode 100644 index 000000000..5b2dec8c6 --- /dev/null +++ b/appc-dispatcher/appc-dispatcher-common/execution-queue-management-lib/src/test/java/org/onap/appc/executionqueue/impl/QueueManagerTest.java @@ -0,0 +1,68 @@ +/* + * ============LICENSE_START======================================================= + * ONAP : APPC + * ================================================================================ + * Copyright (C) 2019 Ericsson + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * ============LICENSE_END========================================================= + */ + +package org.onap.appc.executionqueue.impl; + +import org.junit.Test; +import static org.junit.Assert.assertEquals; +import java.util.ArrayList; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import org.mockito.Mockito; +import org.onap.appc.executionqueue.helper.Util; +import org.powermock.reflect.Whitebox; + +public class QueueManagerTest { + + @Test(expected = NullPointerException.class) + public void testInit() { + QueueManager qm = new QueueManager(); + Util util = Mockito.mock(Util.class); + Mockito.when(util.getExecutionQueueSize()).thenReturn(1); + Mockito.when(util.getThreadPoolSize()).thenReturn(1); + qm.setExecutionQueueUtil(util); + qm.init(); + assertEquals(Integer.valueOf(1), Whitebox.getInternalState(qm, "maxThreadSize")); + } + + @Test + public void testStop() throws InterruptedException { + QueueManager qm = Mockito.spy(new QueueManager()); + ExecutorService executor = Mockito.mock(ExecutorService.class); + Mockito.when(executor.shutdownNow()).thenReturn(new ArrayList<Runnable>()); + Mockito.when(executor.awaitTermination(100, TimeUnit.MILLISECONDS)).thenReturn(false).thenReturn(true); + Whitebox.setInternalState(qm, "messageExecutor", executor); + qm.stop(); + Mockito.verify(executor, Mockito.times(2)).awaitTermination(100, TimeUnit.MILLISECONDS); + } + + @Test + public void testEnqueueTask() throws InterruptedException { + QueueManager qm = Mockito.spy(new QueueManager()); + ExecutorService executor = Mockito.mock(ExecutorService.class); + Mockito.when(executor.shutdownNow()).thenReturn(new ArrayList<Runnable>()); + Mockito.when(executor.awaitTermination(100, TimeUnit.MILLISECONDS)).thenReturn(false).thenReturn(true); + Whitebox.setInternalState(qm, "messageExecutor", executor); + qm.enqueueTask(null); + Mockito.verify(executor).execute(Mockito.any(Runnable.class)); + } + +} |