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.onap.vfc.nfvo.emsdriver.commons.constant.Constant;
  19 import org.onap.vfc.nfvo.emsdriver.commons.model.CollectMsg;
  20 import org.onap.vfc.nfvo.emsdriver.commons.utils.DriverThread;
  21 import org.onap.vfc.nfvo.emsdriver.messagemgr.MessageChannel;
  22 import org.onap.vfc.nfvo.emsdriver.messagemgr.MessageChannelFactory;
  23
  24 public class CollectMsgReceiverThread extends DriverThread{
  25
  26         private long timeStamp = System.currentTimeMillis();
  27
  28         private MessageChannel collectChannel;
  29
  30         private TaskThreadService taskService;
  31
  32         private int thread_max_num = 100;
  33
  34
  35
  36         @Override
  37         public void dispose() {
  38                 collectChannel =  MessageChannelFactory.getMessageChannel(Constant.COLLECT_CHANNEL_KEY);
  39
  40                 taskService = TaskThreadService.getInstance(thread_max_num);
  41                 taskService.start();
  42
  43                 while (isRun()) {
  44
  45                         try {
  46                                 if(System.currentTimeMillis() - timeStamp > Constant.ONEMINUTE){
  47                                         timeStamp = System.currentTimeMillis();
  48
  49                                         log.debug("COLLECT_CHANNEL Msg size :"+collectChannel.size());
  50                                 }
  51
  52                                 Object obj = collectChannel.poll();
  53                                 if(obj == null){
  54                                         continue;
  55                                 }
  56                                 if(obj != null && obj instanceof CollectMsg){
  57                                         CollectMsg collectMsg = (CollectMsg)obj;
  58                                         taskService.add(collectMsg);
  59                                         log.debug("receive a CollectMsg id = "+collectMsg.getId());
  60                                 }else{
  61                                         log.error("receive Objcet not CollectMsg "+obj);
  62                                 }
  63
  64                         } catch (Exception e) {
  65                                 log.error("dispatch alarm exception",e);
  66
  67                         }
  68                 }
  69
  70         }
  71
  72
  73
  74         /**
  75          * @return the thread_max_num
  76          */
  77         public int getThread_max_num() {
  78                 return thread_max_num;
  79         }
  80
  81
  82
  83         /**
  84          * @param thread_max_num the thread_max_num to set
  85          */
  86         public void setThread_max_num(int thread_max_num) {
  87                 this.thread_max_num = thread_max_num;
  88         }
  89
  90
  91 }