From b464eec16a30627bf5237b71ba0088d458640c22 Mon Sep 17 00:00:00 2001 From: FengLiang Date: Mon, 6 Mar 2017 15:04:36 +0800 Subject: Modify MQ listener mode Change-Id: Ided322b55ed1baff351edb03239fc06e32da3844 Issue-ID: HOLMES-47 Signed-off-by: FengLiang --- .../openo/holmes/engine/manager/DroolsEngine.java | 78 ++++++++++------ .../holmes/engine/manager/DroolsEngineTest.java | 102 +++++++++++++++++---- 2 files changed, 133 insertions(+), 47 deletions(-) (limited to 'engine-d/src') diff --git a/engine-d/src/main/java/org/openo/holmes/engine/manager/DroolsEngine.java b/engine-d/src/main/java/org/openo/holmes/engine/manager/DroolsEngine.java index 0ed452a..d9f55ed 100644 --- a/engine-d/src/main/java/org/openo/holmes/engine/manager/DroolsEngine.java +++ b/engine-d/src/main/java/org/openo/holmes/engine/manager/DroolsEngine.java @@ -16,6 +16,7 @@ package org.openo.holmes.engine.manager; +import java.io.Serializable; import java.io.StringReader; import java.util.List; import java.util.Locale; @@ -25,11 +26,13 @@ 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.ObjectMessage; +import javax.jms.MessageListener; import javax.jms.Session; import lombok.extern.slf4j.Slf4j; import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.command.ActiveMQObjectMessage; import org.drools.KnowledgeBase; import org.drools.KnowledgeBaseConfiguration; import org.drools.KnowledgeBaseFactory; @@ -57,11 +60,6 @@ import org.openo.holmes.engine.wrapper.RuleMgtWrapper; @Slf4j @Service public class DroolsEngine { - - private final static String CORRELATION_RULE = "CORRELATION_RULE"; - - private final static String CORRELATION_ALARM = "CORRELATION_ALARM"; - private final static int ENABLE = 1; @Inject @@ -99,8 +97,8 @@ public class DroolsEngine { connectionFactory = new ActiveMQConnectionFactory(mqConfigProvider.get().brokerUsername, mqConfigProvider.get().brokerPassword, brokerURL); - Thread thread = new Thread(new AlarmMqMessageListener()); - thread.start(); + AlarmMqMessageListener listener = new AlarmMqMessageListener(); + listener.receive(); } @@ -258,34 +256,56 @@ public class DroolsEngine { this.ksession.fireAllRules(); } - class AlarmMqMessageListener implements Runnable { + class AlarmMqMessageListener implements MessageListener { + + private Connection connection = null; + private Session session = null; + private Destination destination = null; + private MessageConsumer consumer = null; - public void run() { - Connection connection; - Session session; - Destination destination; - MessageConsumer messageConsumer; + 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 { - connection = connectionFactory.createConnection(); - connection.start(); - session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); - destination = session.createTopic(AlarmConst.MQ_TOPIC_NAME_ALARM); - messageConsumer = session.createConsumer(destination); - - while (true) { - ObjectMessage objMessage = (ObjectMessage) messageConsumer.receive(100000); - if (objMessage != null) { - putRaisedIntoStream((Alarm) objMessage.getObject()); - } else { - break; - } + Serializable object = objectMessage.getObject(); + + if (object instanceof Alarm) { + Alarm alarm = (Alarm) object; + putRaisedIntoStream(alarm); } } catch (JMSException e) { - log.error("connection mq service Failed: " + e.getMessage(), 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(); } } - } diff --git a/engine-d/src/test/java/org/openo/holmes/engine/manager/DroolsEngineTest.java b/engine-d/src/test/java/org/openo/holmes/engine/manager/DroolsEngineTest.java index 1740d62..0f637c7 100644 --- a/engine-d/src/test/java/org/openo/holmes/engine/manager/DroolsEngineTest.java +++ b/engine-d/src/test/java/org/openo/holmes/engine/manager/DroolsEngineTest.java @@ -36,6 +36,7 @@ import javax.jms.MessageConsumer; import javax.jms.ObjectMessage; import javax.jms.Session; import javax.jms.Topic; +import org.apache.activemq.command.ActiveMQObjectMessage; import org.drools.KnowledgeBase; import org.drools.KnowledgeBaseConfiguration; import org.drools.builder.KnowledgeBuilder; @@ -362,59 +363,124 @@ public class DroolsEngineTest { PowerMock.verifyAll(); } + @Test - public void listener_run_objmessage_is_null() throws JMSException { + public void listener_receive() throws JMSException { DroolsEngine.AlarmMqMessageListener listener = droolsEngine.new AlarmMqMessageListener(); Connection connection = PowerMock.createMock(Connection.class); Session session = PowerMock.createMock(Session.class); Destination destination = PowerMock.createMock(Topic.class); - MessageConsumer messageConsumer = PowerMock.createMock(MessageConsumer.class); + MessageConsumer consumer = PowerMock.createMock(MessageConsumer.class); + + Whitebox.setInternalState(listener, "connection", connection); + Whitebox.setInternalState(listener, "session", session); + Whitebox.setInternalState(listener, "destination", destination); + Whitebox.setInternalState(listener, "consumer", consumer); + + PowerMock.reset(); expect(connectionFactory.createConnection()).andReturn(connection); connection.start(); expect(connection.createSession(anyBoolean(), anyInt())).andReturn(session); expect(session.createTopic(anyObject(String.class))).andReturn((Topic) destination); - expect(session.createConsumer(anyObject(Destination.class))).andReturn(messageConsumer); - expect(messageConsumer.receive(anyLong())).andReturn(null); + expect(session.createConsumer(anyObject(Destination.class))).andReturn(consumer); + consumer.setMessageListener(listener); PowerMock.replayAll(); - listener.run(); + listener.receive(); PowerMock.verifyAll(); } @Test - public void listener_run_objmessage_is_not_null() throws JMSException { + public void listener_exception() throws JMSException { DroolsEngine.AlarmMqMessageListener listener = droolsEngine.new AlarmMqMessageListener(); Connection connection = PowerMock.createMock(Connection.class); Session session = PowerMock.createMock(Session.class); Destination destination = PowerMock.createMock(Topic.class); - MessageConsumer messageConsumer = PowerMock.createMock(MessageConsumer.class); - ObjectMessage objMessage = PowerMock.createMock(ObjectMessage.class); + MessageConsumer consumer = PowerMock.createMock(MessageConsumer.class); - FactHandle factHandle = PowerMock.createMock(FactHandle.class); + Whitebox.setInternalState(listener, "connection", connection); + Whitebox.setInternalState(listener, "session", session); + Whitebox.setInternalState(listener, "destination", destination); + Whitebox.setInternalState(listener, "consumer", consumer); + + PowerMock.reset(); expect(connectionFactory.createConnection()).andReturn(connection); connection.start(); expect(connection.createSession(anyBoolean(), anyInt())).andReturn(session); expect(session.createTopic(anyObject(String.class))).andReturn((Topic) destination); - expect(session.createConsumer(anyObject(Destination.class))).andReturn(messageConsumer); - expect(messageConsumer.receive(anyLong())).andReturn(objMessage); - expect(objMessage.getObject()).andReturn(new Alarm()); + expect(session.createConsumer(anyObject(Destination.class))).andReturn(consumer); + consumer.setMessageListener(listener); + EasyMock.expectLastCall().andThrow(new JMSException("")); - expect(ksession.getFactHandle(anyObject(Alarm.class))).andReturn(factHandle); - ksession.retract(anyObject(FactHandle.class)); - expect(ksession.insert(anyObject(Alarm.class))).andReturn(null); - expect(ksession.fireAllRules()).andReturn(0); + consumer.close(); + session.close(); + connection.close(); + + + PowerMock.replayAll(); + + listener.receive(); + + PowerMock.verifyAll(); + } + + @Test + public void listener_close_exception() throws JMSException { + DroolsEngine.AlarmMqMessageListener listener = droolsEngine.new AlarmMqMessageListener(); + + Connection connection = PowerMock.createMock(Connection.class); + Session session = PowerMock.createMock(Session.class); + Destination destination = PowerMock.createMock(Topic.class); + MessageConsumer consumer = PowerMock.createMock(MessageConsumer.class); - expect(messageConsumer.receive(anyLong())).andReturn(null); + Whitebox.setInternalState(listener, "connection", connection); + Whitebox.setInternalState(listener, "session", session); + Whitebox.setInternalState(listener, "destination", destination); + Whitebox.setInternalState(listener, "consumer", consumer); + + PowerMock.reset(); + + expect(connectionFactory.createConnection()).andReturn(connection); + connection.start(); + expect(connection.createSession(anyBoolean(), anyInt())).andReturn(session); + expect(session.createTopic(anyObject(String.class))).andReturn((Topic) destination); + expect(session.createConsumer(anyObject(Destination.class))).andReturn(consumer); + consumer.setMessageListener(listener); + EasyMock.expectLastCall().andThrow(new JMSException("")); + + consumer.close(); + EasyMock.expectLastCall().andThrow(new JMSException("")); + + + PowerMock.replayAll(); + + listener.receive(); + + PowerMock.verifyAll(); + } + + @Test + public void listener_on_message() throws JMSException { + DroolsEngine.AlarmMqMessageListener listener = droolsEngine.new AlarmMqMessageListener(); + Alarm alarm = new Alarm(); + alarm.setAlarmKey("alarmKey"); + ActiveMQObjectMessage objectMessage = new ActiveMQObjectMessage(); + objectMessage.setObject(alarm); + + expect(ksession.getFactHandle(anyObject(Alarm.class))).andReturn(null); + + expect(ksession.insert(anyObject(Alarm.class))).andReturn(null); + expect(ksession.fireAllRules()).andReturn(1); PowerMock.replayAll(); - listener.run(); + listener.onMessage(objectMessage); PowerMock.verifyAll(); } -- cgit 1.2.3-korg