summaryrefslogtreecommitdiffstats
path: root/engine-d/src/main/java
diff options
context:
space:
mode:
authorShiwei Tian <tian.shiwei@zte.com.cn>2017-09-12 19:06:49 +0800
committerShiwei Tian <tian.shiwei@zte.com.cn>2017-09-12 19:06:49 +0800
commit9205cee30a28456ca29918a9c2036e51a2feb301 (patch)
treeb771afaaef20a53980831245ea03e710016daf84 /engine-d/src/main/java
parent3c2a7c4614ed0b925a7ac9d0c2ff4907c89e6b51 (diff)
add MQ consumer
Issue-ID: HOLMES-47 Change-Id: I16109e668ca20819af2e53de906387ec12907e7b Signed-off-by: Shiwei Tian <tian.shiwei@zte.com.cn>
Diffstat (limited to 'engine-d/src/main/java')
-rw-r--r--engine-d/src/main/java/org/onap/holmes/engine/mqconsumer/MQConsumer.java114
1 files changed, 114 insertions, 0 deletions
diff --git a/engine-d/src/main/java/org/onap/holmes/engine/mqconsumer/MQConsumer.java b/engine-d/src/main/java/org/onap/holmes/engine/mqconsumer/MQConsumer.java
new file mode 100644
index 0000000..bd77312
--- /dev/null
+++ b/engine-d/src/main/java/org/onap/holmes/engine/mqconsumer/MQConsumer.java
@@ -0,0 +1,114 @@
+/**
+ * Copyright 2017 ZTE Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onap.holmes.engine.mqconsumer;
+
+import java.io.Serializable;
+import javax.inject.Inject;
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.Session;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.command.ActiveMQObjectMessage;
+import org.glassfish.hk2.api.IterableProvider;
+import org.jvnet.hk2.annotations.Service;
+import org.onap.holmes.common.api.stat.VesAlarm;
+import org.onap.holmes.common.config.MQConfig;
+import org.onap.holmes.common.constant.AlarmConst;
+import org.onap.holmes.engine.manager.DroolsEngine;
+
+@Service
+@Slf4j
+@NoArgsConstructor
+public class MQConsumer {
+
+ @Inject
+ private IterableProvider<MQConfig> mqConfigProvider;
+ private ConnectionFactory connectionFactory;
+ private ConnectionFactory connectionFactory1;
+ @Inject
+ private DroolsEngine engine;
+
+ public void registerAlarmTopicListener() {
+ String brokerURL =
+ "tcp://" + mqConfigProvider.get().brokerIp + ":" + mqConfigProvider.get().brokerPort;
+ connectionFactory = new ActiveMQConnectionFactory(mqConfigProvider.get().brokerUsername,
+ mqConfigProvider.get().brokerPassword, brokerURL);
+
+ AlarmMqMessageListener listener = new AlarmMqMessageListener();
+ listener.receive();
+ }
+ class AlarmMqMessageListener implements MessageListener {
+
+ private Connection connection = null;
+ private Session session = null;
+ private Destination destination = null;
+ private MessageConsumer consumer = null;
+
+ private void initialize() throws JMSException {
+ connection = connectionFactory.createConnection();
+ session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ destination = session.createTopic(AlarmConst.MQ_TOPIC_NAME_ALARM);
+ consumer = session.createConsumer(destination);
+ connection.start();
+ }
+
+ public void receive() {
+ try {
+ initialize();
+ consumer.setMessageListener(this);
+ } catch (JMSException e) {
+ log.error("Failed to connect to the MQ service : " + e.getMessage(), e);
+ try {
+ close();
+ } catch (JMSException e1) {
+ log.error("Failed close connection " + e1.getMessage(), e1);
+ }
+ }
+ }
+
+ public void onMessage(Message arg0) {
+ ActiveMQObjectMessage objectMessage = (ActiveMQObjectMessage) arg0;
+ try {
+ Serializable object = objectMessage.getObject();
+ if (object instanceof VesAlarm) {
+ VesAlarm vesAlarm = (VesAlarm) object;
+ engine.putRaisedIntoStream(vesAlarm);
+ }
+ } catch (JMSException e) {
+ log.error("Failed get object : " + e.getMessage(), e);
+ }
+ }
+
+ private void close() throws JMSException {
+ if (consumer != null) {
+ consumer.close();
+ }
+ if (session != null) {
+ session.close();
+ }
+ if (connection != null) {
+ connection.close();
+ }
+ }
+ }
+}