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                                         Thread.sleep(10);
  55                                         continue;
  56                                 }
  57                                 if(obj != null && obj instanceof CollectMsg){
  58                                         CollectMsg collectMsg = (CollectMsg)obj;
  59                                         taskService.add(collectMsg);
  60                                         log.debug("receive a CollectMsg id = "+collectMsg.getId());
  61                                 }else{
  62                                         log.error("receive Objcet not CollectMsg "+obj);
  63                                 }
  64
  65                         } catch (Exception e) {
  66                                 log.error("dispatch alarm exception",e);
  67
  68                         }
  69                 }
  70
  71         }
  72
  73
  74
  75         /**
  76          * @return the thread_max_num
  77          */
  78         public int getThread_max_num() {
  79                 return thread_max_num;
  80         }
  81
  82
  83
  84         /**
  85          * @param thread_max_num the thread_max_num to set
  86          */
  87         public void setThread_max_num(int thread_max_num) {
  88                 this.thread_max_num = thread_max_num;
  89         }
  90
  91
  92 }