2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright (C) 2017 AT&T Intellectual Property. All rights
7 * ================================================================================
8 * Licensed under the Apache License, Version 2.0 (the "License");
9 * you may not use this file except in compliance with the License.
10 * You may obtain a copy of the License at
12 * http://www.apache.org/licenses/LICENSE-2.0
14 * Unless required by applicable law or agreed to in writing, software
15 * distributed under the License is distributed on an "AS IS" BASIS,
16 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17 * See the License for the specific language governing permissions and
18 * limitations under the License.
19 * ============LICENSE_END=========================================================
22 package org.openecomp.appc.executionqueue.impl;
24 import java.util.concurrent.ExecutorService;
25 import java.util.concurrent.Executors;
26 import java.util.concurrent.LinkedBlockingQueue;
28 import org.openecomp.appc.executionqueue.MessageExpirationListener;
29 import org.openecomp.appc.executionqueue.helper.Util;
30 import org.openecomp.appc.executionqueue.impl.object.QueueMessage;
31 import com.att.eelf.configuration.EELFLogger;
32 import com.att.eelf.configuration.EELFManager;
34 public class QueueManager {
36 private LinkedBlockingQueue<QueueMessage> queue;
38 private MessageExpirationListener listener;
40 private static int MAX_QUEUE_SIZE = Util.getExecutionQueSize();
42 private static int MAX_THREAD_SIZE = Util.getThreadPoolSize();
44 private ExecutorService messageExecutor;
46 private static final EELFLogger logger =
47 EELFManager.getInstance().getLogger(QueueManager.class);
49 private QueueManager(){
53 private static class QueueManagerHolder {
54 private static final QueueManager INSTANCE = new QueueManager();
57 public static QueueManager getInstance() {
58 return QueueManagerHolder.INSTANCE;
62 queue = new LinkedBlockingQueue(MAX_QUEUE_SIZE);
63 messageExecutor = Executors.newFixedThreadPool(MAX_THREAD_SIZE,Util.getThreadFactory(true));
65 for(int i=0;i<MAX_THREAD_SIZE;i++){
66 messageExecutor.submit(new Runnable() {
71 QueueMessage queueMessage = queue.take();
72 if(messageExpired(queueMessage)){
73 logger.debug("Message expired "+ queueMessage.getMessage());
75 listener.onMessageExpiration(queueMessage.getMessage());
78 logger.warn("Listener not available for expired message ");
82 queueMessage.getMessage().run();
84 } catch (Exception e) {
85 logger.error("Error in startMessagePolling method of ExecutionQueueServiceImpl" + e.getMessage());
93 public void setListener(MessageExpirationListener listener) {
94 this.listener = listener;
97 public boolean enqueueTask(QueueMessage queueMessage) {
98 return queue.offer(queueMessage);
101 private boolean messageExpired(QueueMessage queueMessage) {
102 if(queueMessage.getExpirationTime() != null){
103 return queueMessage.getExpirationTime().getTime() < System.currentTimeMillis();