2 * ============LICENSE_START=======================================================
3 * Copyright (C) 2016-2018 Ericsson. All rights reserved.
4 * Modifications Copyright (C) 2019-2020 Nordix Foundation.
5 * ================================================================================
6 * Licensed under the Apache License, Version 2.0 (the "License");
7 * you may not use this file except in compliance with the License.
8 * You may obtain a copy of the License at
10 * http://www.apache.org/licenses/LICENSE-2.0
12 * Unless required by applicable law or agreed to in writing, software
13 * distributed under the License is distributed on an "AS IS" BASIS,
14 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15 * See the License for the specific language governing permissions and
16 * limitations under the License.
18 * SPDX-License-Identifier: Apache-2.0
19 * ============LICENSE_END=========================================================
22 package org.onap.policy.apex.plugins.event.carrier.jms;
24 import java.util.Properties;
25 import javax.jms.Connection;
26 import javax.jms.ConnectionFactory;
27 import javax.jms.Message;
28 import javax.jms.MessageConsumer;
29 import javax.jms.MessageListener;
30 import javax.jms.Session;
31 import javax.jms.Topic;
32 import javax.naming.InitialContext;
33 import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities;
34 import org.onap.policy.apex.service.engine.event.ApexEventException;
35 import org.onap.policy.apex.service.engine.event.ApexEventReceiver;
36 import org.onap.policy.apex.service.engine.event.ApexEventRuntimeException;
37 import org.onap.policy.apex.service.engine.event.ApexPluginsEventConsumer;
38 import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerParameters;
39 import org.slf4j.Logger;
40 import org.slf4j.LoggerFactory;
43 * This class implements an Apex event consumer that receives events using JMS.
45 * @author Liam Fallon (liam.fallon@ericsson.com)
47 public class ApexJmsConsumer extends ApexPluginsEventConsumer implements MessageListener {
48 // Get a reference to the logger
49 private static final Logger LOGGER = LoggerFactory.getLogger(ApexJmsConsumer.class);
51 // The Apex and JMS parameters read from the parameter service
52 private JmsCarrierTechnologyParameters jmsConsumerProperties;
54 // The event receiver that will receive events from this consumer
55 private ApexEventReceiver eventReceiver;
57 // The connection to the JMS server
58 private Connection connection;
60 // The topic on which we receive events from JMS
61 private Topic jmsIncomingTopic;
64 public void init(final String consumerName, final EventHandlerParameters consumerParameters,
65 final ApexEventReceiver incomingEventReceiver) throws ApexEventException {
66 this.eventReceiver = incomingEventReceiver;
68 this.name = consumerName;
70 // Check and get the JMS Properties
71 if (!(consumerParameters.getCarrierTechnologyParameters() instanceof JmsCarrierTechnologyParameters)) {
72 final String errorMessage = "specified consumer properties of type \""
73 + consumerParameters.getCarrierTechnologyParameters().getClass().getName()
74 + "\" are not applicable to a JMS consumer";
75 throw new ApexEventException(errorMessage);
77 jmsConsumerProperties = (JmsCarrierTechnologyParameters) consumerParameters.getCarrierTechnologyParameters();
79 // Look up the JMS connection factory
80 InitialContext jmsContext = null;
81 ConnectionFactory connectionFactory = null;
83 jmsContext = new InitialContext(jmsConsumerProperties.getJmsConsumerProperties());
84 connectionFactory = (ConnectionFactory) jmsContext.lookup(jmsConsumerProperties.getConnectionFactory());
86 // Check if we actually got a connection factory
87 if (connectionFactory == null) {
88 throw new IllegalArgumentException(
89 "JMS context lookup of \"" + jmsConsumerProperties.getConnectionFactory() + "\" returned null");
91 } catch (final Exception e) {
92 final String errorMessage = "lookup of JMS connection factory \""
93 + jmsConsumerProperties.getConnectionFactory() + "\" failed for JMS consumer properties \""
94 + jmsConsumerProperties.getJmsConsumerProperties() + "\"";
95 throw new ApexEventException(errorMessage, e);
98 // Lookup the topic on which we will receive events
100 jmsIncomingTopic = (Topic) jmsContext.lookup(jmsConsumerProperties.getConsumerTopic());
102 // Check if we actually got a topic
103 if (jmsIncomingTopic == null) {
104 throw new IllegalArgumentException(
105 "JMS context lookup of \"" + jmsConsumerProperties.getConsumerTopic() + "\" returned null");
107 } catch (final Exception e) {
108 final String errorMessage = "lookup of JMS topic \"" + jmsConsumerProperties.getConsumerTopic()
109 + "\" failed for JMS consumer properties \"" + jmsConsumerProperties.getJmsConsumerProperties()
111 throw new ApexEventException(errorMessage, e);
114 // Create and start a connection to the JMS server
116 connection = connectionFactory.createConnection(jmsConsumerProperties.getSecurityPrincipal(),
117 jmsConsumerProperties.getSecurityCredentials());
119 } catch (final Exception e) {
120 final String errorMessage = "connection to the JMS server failed for JMS properties \""
121 + jmsConsumerProperties.getJmsConsumerProperties() + "\"";
122 throw new ApexEventException(errorMessage, e);
131 // JMS session and message consumer for receiving messages
132 try (final Session jmsSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)) {
133 // Create a message consumer for reception of messages and set this class as a message listener
134 createMessageConsumer(jmsSession);
135 } catch (final Exception e) {
136 final String errorMessage = "failed to create a JMS session towards the JMS server for receiving messages";
137 LOGGER.warn(errorMessage, e);
138 throw new ApexEventRuntimeException(errorMessage, e);
140 // Everything is now set up
141 if (LOGGER.isDebugEnabled()) {
142 LOGGER.debug("event receiver " + this.getClass().getName() + ":" + this.name + " subscribed to JMS topic: "
143 + jmsConsumerProperties.getConsumerTopic());
148 * The helper function to create a message consumer from a given JMS session.
150 * @param jmsSession a JMS session
152 private void createMessageConsumer(final Session jmsSession) {
153 try (final MessageConsumer messageConsumer = jmsSession.createConsumer(jmsIncomingTopic)) {
154 messageConsumer.setMessageListener(this);
156 // The endless loop that receives events over JMS
157 while (consumerThread.isAlive() && !stopOrderedFlag) {
158 ThreadUtilities.sleep(jmsConsumerProperties.getConsumerWaitTime());
160 } catch (final Exception e) {
161 final String errorMessage = "failed to create a JMS message consumer for receiving messages";
162 LOGGER.warn(errorMessage, e);
163 throw new ApexEventRuntimeException(errorMessage, e);
171 public void onMessage(final Message jmsMessage) {
173 if (LOGGER.isTraceEnabled()) {
174 LOGGER.trace("event received for {} for forwarding to Apex engine : {} {}",
175 this.getClass().getName() + ":" + this.name, jmsMessage.getJMSMessageID(),
176 jmsMessage.getJMSType());
179 eventReceiver.receiveEvent(new Properties(), jmsMessage);
180 } catch (final Exception e) {
181 final String errorMessage = "failed to receive message from JMS";
182 LOGGER.warn(errorMessage, e);
183 throw new ApexEventRuntimeException(errorMessage, e);
192 stopOrderedFlag = true;
194 while (consumerThread.isAlive()) {
195 ThreadUtilities.sleep(jmsConsumerProperties.getConsumerWaitTime());
198 // Close the connection to the JMS server
200 if (connection != null) {
203 } catch (final Exception e) {
204 final String errorMessage = "close of connection to the JMS server failed";
205 LOGGER.warn(errorMessage, e);