ems/boco/src/main/java/org/onap/vfc/nfvo/emsdriver/collector/CollectMsgReceiverThread.java

   1 /*
   2  * Copyright 2017 BOCO Corporation.  CMCC Technologies Co., Ltd
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16 package org.onap.vfc.nfvo.emsdriver.collector;
  17
  18 import org.slf4j.Logger;
  19 import org.slf4j.LoggerFactory;
  20 import org.onap.vfc.nfvo.emsdriver.commons.constant.Constant;
  21 import org.onap.vfc.nfvo.emsdriver.commons.model.CollectMsg;
  22 import org.onap.vfc.nfvo.emsdriver.commons.utils.DriverThread;
  23 import org.onap.vfc.nfvo.emsdriver.configmgr.ConfigurationManager;
  24 import org.onap.vfc.nfvo.emsdriver.messagemgr.MessageChannel;
  25 import org.onap.vfc.nfvo.emsdriver.messagemgr.MessageChannelFactory;
  26
  27 public class CollectMsgReceiverThread extends DriverThread {
  28         protected static final Logger logger = LoggerFactory.getLogger(CollectMsgReceiverThread.class);
  29
  30         private long timeStamp = System.currentTimeMillis();
  31
  32         private MessageChannel collectChannel;
  33
  34         private TaskThreadService taskService;
  35
  36         private int threadMaxNum = 100;
  37
  38         @Override
  39         public void dispose() {
  40                 collectChannel = MessageChannelFactory
  41                                 .getMessageChannel(Constant.COLLECT_CHANNEL_KEY);
  42                 taskService = TaskThreadService.getInstance(threadMaxNum);
  43                 taskService.start();
  44
  45                 while (isRun()) {
  46
  47                         try {
  48                                 if (System.currentTimeMillis() - timeStamp > Constant.ONEMINUTE) {
  49                                         timeStamp = System.currentTimeMillis();
  50                                         logger.debug("COLLECT_CHANNEL Msg size :"
  51                                                         + collectChannel.size());
  52                                 }
  53                                 Object obj = collectChannel.poll();
  54                                 if (obj == null) {
  55                                         Thread.sleep(10);
  56                                         continue;
  57                                 }
  58                                 if (obj instanceof CollectMsg) {
  59                                         CollectMsg collectMsg = (CollectMsg) obj;
  60                                         taskService.add(collectMsg);
  61                                         logger.debug("receive a CollectMsg id = " + collectMsg.getId());
  62                                 } else {
  63                                         logger.error("receive Objcet not CollectMsg " + obj);
  64                                 }
  65
  66                         } catch (Exception e) {
  67                                 logger.error("dispatch alarm exception", e);
  68
  69                         }
  70                 }
  71
  72         }
  73
  74         /**
  75          * @return the threadMaxNum
  76          */
  77         public int getThreadMaxNum() {
  78                 return threadMaxNum;
  79         }
  80
  81         /**
  82          * @param threadMaxNum  the threadMaxNum to set
  83          */
  84         public void setThreadMaxNum(int threadMaxNum) {
  85                 this.threadMaxNum = threadMaxNum;
  86         }
  87
  88         /**
  89          * @return the taskService
  90          */
  91         public TaskThreadService getTaskService() {
  92                 return taskService;
  93         }
  94
  95 }