holmes-actions/src/main/java/org/openo/holmes/common/producer/MQProducer.java

   1 /**
   2  * Copyright 2017 ZTE Corporation.
   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.openo.holmes.common.producer;
  17
  18 import javax.annotation.PostConstruct;
  19 import javax.inject.Inject;
  20 import javax.jms.Connection;
  21 import javax.jms.ConnectionFactory;
  22 import javax.jms.Destination;
  23 import javax.jms.JMSException;
  24 import javax.jms.MessageProducer;
  25 import javax.jms.ObjectMessage;
  26 import javax.jms.Session;
  27 import lombok.extern.slf4j.Slf4j;
  28 import org.apache.activemq.ActiveMQConnectionFactory;
  29 import org.glassfish.hk2.api.IterableProvider;
  30 import org.jvnet.hk2.annotations.Service;
  31 import org.openo.holmes.common.api.entity.CorrelationResult;
  32 import org.openo.holmes.common.api.stat.Alarm;
  33 import org.openo.holmes.common.api.stat.AplusResult;
  34 import org.openo.holmes.common.config.MQConfig;
  35 import org.openo.holmes.common.constant.AlarmConst;
  36
  37 @Service
  38 @Slf4j
  39 public class MQProducer {
  40
  41     @Inject
  42     private IterableProvider<MQConfig> mqConfigProvider;
  43     private ConnectionFactory connectionFactory;
  44
  45     @PostConstruct
  46     public void init() {
  47
  48         String brokerURL =
  49             "tcp://" + mqConfigProvider.get().brokerIp + ":" + mqConfigProvider.get().brokerPort;
  50         connectionFactory = new ActiveMQConnectionFactory(mqConfigProvider.get().brokerUsername,
  51             mqConfigProvider.get().brokerPassword, brokerURL);
  52     }
  53
  54     public void sendAlarmMQTopicMsg(Alarm alarm) {
  55
  56         Connection connection = null;
  57         Session session;
  58         Destination destination;
  59         MessageProducer messageProducer;
  60
  61         try {
  62
  63             connection = connectionFactory.createConnection();
  64             connection.start();
  65             session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
  66             destination = session.createTopic(AlarmConst.MQ_TOPIC_NAME_ALARM);
  67             messageProducer = session.createProducer(destination);
  68             ObjectMessage message = session.createObjectMessage(alarm);
  69             messageProducer.send(message);
  70             session.commit();
  71
  72         } catch (Exception e) {
  73             log.error("Failed send alarm." + e.getMessage(), e);
  74         } finally {
  75             if (connection != null) {
  76                 try {
  77                     connection.close();
  78                 } catch (JMSException e) {
  79                     log.error("Failed close connection." + e.getMessage(), e);
  80                 }
  81             }
  82         }
  83     }
  84
  85     public void sendCorrelationMQTopicMsg(String ruleId, long createTimeL, Alarm parentAlarm,
  86         Alarm childAlarm) {
  87
  88         CorrelationResult correlationResult = new CorrelationResult();
  89         correlationResult.setRuleId(ruleId);
  90         correlationResult.setCreateTimeL(createTimeL);
  91         correlationResult.setResultType(AplusResult.APLUS_CORRELATION);
  92         correlationResult.setAffectedAlarms(new Alarm[]{parentAlarm, childAlarm});
  93
  94         Connection connection = null;
  95         Session session;
  96         Destination destination;
  97         MessageProducer messageProducer;
  98
  99         try {
 100
 101             connection = connectionFactory.createConnection();
 102             connection.start();
 103             session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
 104             destination = session.createTopic(AlarmConst.MQ_TOPIC_NAME_ALARMS_CORRELATION);
 105             messageProducer = session.createProducer(destination);
 106             ObjectMessage message = session.createObjectMessage(correlationResult);
 107             messageProducer.send(message);
 108             session.commit();
 109
 110         } catch (Exception e) {
 111             log.error("Failed send correlation." + e.getMessage(), e);
 112         } finally {
 113             if (connection != null) {
 114                 try {
 115                     connection.close();
 116                 } catch (JMSException e) {
 117                     log.error("Failed close connection." + e.getMessage(), e);
 118                 }
 119             }
 120         }
 121     }
 122 }