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;
26 import javax.jms.Connection;
27 import javax.jms.ConnectionFactory;
28 import javax.jms.Message;
29 import javax.jms.MessageConsumer;
30 import javax.jms.MessageListener;
31 import javax.jms.Session;
32 import javax.jms.Topic;
33 import javax.naming.InitialContext;
35 import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities;
36 import org.onap.policy.apex.service.engine.event.ApexEventException;
37 import org.onap.policy.apex.service.engine.event.ApexEventReceiver;
38 import org.onap.policy.apex.service.engine.event.ApexEventRuntimeException;
39 import org.onap.policy.apex.service.engine.event.ApexPluginsEventConsumer;
40 import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerParameters;
41 import org.slf4j.Logger;
42 import org.slf4j.LoggerFactory;
45 * This class implements an Apex event consumer that receives events using JMS.
47 * @author Liam Fallon (liam.fallon@ericsson.com)
49 public class ApexJmsConsumer extends ApexPluginsEventConsumer implements MessageListener {
50 // Get a reference to the logger
51 private static final Logger LOGGER = LoggerFactory.getLogger(ApexJmsConsumer.class);
53 // The Apex and JMS parameters read from the parameter service
54 private JmsCarrierTechnologyParameters jmsConsumerProperties;
56 // The event receiver that will receive events from this consumer
57 private ApexEventReceiver eventReceiver;
59 // The connection to the JMS server
60 private Connection connection;
62 // The topic on which we receive events from JMS
63 private Topic jmsIncomingTopic;
66 public void init(final String consumerName, final EventHandlerParameters consumerParameters,
67 final ApexEventReceiver incomingEventReceiver) throws ApexEventException {
68 this.eventReceiver = incomingEventReceiver;
70 this.name = consumerName;
72 // Check and get the JMS Properties
73 if (!(consumerParameters.getCarrierTechnologyParameters() instanceof JmsCarrierTechnologyParameters)) {
74 final String errorMessage = "specified consumer properties of type \""
75 + consumerParameters.getCarrierTechnologyParameters().getClass().getName()
76 + "\" are not applicable to a JMS consumer";
77 throw new ApexEventException(errorMessage);
79 jmsConsumerProperties = (JmsCarrierTechnologyParameters) consumerParameters.getCarrierTechnologyParameters();
81 // Look up the JMS connection factory
82 InitialContext jmsContext = null;
83 ConnectionFactory connectionFactory = null;
85 jmsContext = new InitialContext(jmsConsumerProperties.getJmsConsumerProperties());
86 connectionFactory = (ConnectionFactory) jmsContext.lookup(jmsConsumerProperties.getConnectionFactory());
88 // Check if we actually got a connection factory
89 if (connectionFactory == null) {
90 throw new IllegalArgumentException(
91 "JMS context lookup of \"" + jmsConsumerProperties.getConnectionFactory() + "\" returned null");
93 } catch (final Exception e) {
94 final String errorMessage = "lookup of JMS connection factory \""
95 + jmsConsumerProperties.getConnectionFactory() + "\" failed for JMS consumer properties \""
96 + jmsConsumerProperties.getJmsConsumerProperties() + "\"";
97 throw new ApexEventException(errorMessage, e);
100 // Lookup the topic on which we will receive events
102 jmsIncomingTopic = (Topic) jmsContext.lookup(jmsConsumerProperties.getConsumerTopic());
104 // Check if we actually got a topic
105 if (jmsIncomingTopic == null) {
106 throw new IllegalArgumentException(
107 "JMS context lookup of \"" + jmsConsumerProperties.getConsumerTopic() + "\" returned null");
109 } catch (final Exception e) {
110 final String errorMessage = "lookup of JMS topic \"" + jmsConsumerProperties.getConsumerTopic()
111 + "\" failed for JMS consumer properties \"" + jmsConsumerProperties.getJmsConsumerProperties()
113 throw new ApexEventException(errorMessage, e);
116 // Create and start a connection to the JMS server
118 connection = connectionFactory.createConnection(jmsConsumerProperties.getSecurityPrincipal(),
119 jmsConsumerProperties.getSecurityCredentials());
121 } catch (final Exception e) {
122 final String errorMessage = "connection to the JMS server failed for JMS properties \""
123 + jmsConsumerProperties.getJmsConsumerProperties() + "\"";
124 throw new ApexEventException(errorMessage, e);
133 // JMS session and message consumer for receiving messages
134 try (final Session jmsSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)) {
135 // Create a message consumer for reception of messages and set this class as a message listener
136 createMessageConsumer(jmsSession);
137 } catch (final Exception e) {
138 final String errorMessage = "failed to create a JMS session towards the JMS server for receiving messages";
139 LOGGER.warn(errorMessage, e);
140 throw new ApexEventRuntimeException(errorMessage, e);
142 // Everything is now set up
143 if (LOGGER.isDebugEnabled()) {
144 LOGGER.debug("event receiver " + this.getClass().getName() + ":" + this.name + " subscribed to JMS topic: "
145 + jmsConsumerProperties.getConsumerTopic());
150 * The helper function to create a message consumer from a given JMS session.
152 * @param jmsSession a JMS session
154 private void createMessageConsumer(final Session jmsSession) {
155 try (final MessageConsumer messageConsumer = jmsSession.createConsumer(jmsIncomingTopic)) {
156 messageConsumer.setMessageListener(this);
158 // The endless loop that receives events over JMS
159 while (consumerThread.isAlive() && !stopOrderedFlag) {
160 ThreadUtilities.sleep(jmsConsumerProperties.getConsumerWaitTime());
162 } catch (final Exception e) {
163 final String errorMessage = "failed to create a JMS message consumer for receiving messages";
164 LOGGER.warn(errorMessage, e);
165 throw new ApexEventRuntimeException(errorMessage, e);
173 public void onMessage(final Message jmsMessage) {
175 if (LOGGER.isTraceEnabled()) {
176 LOGGER.trace("event received for {} for forwarding to Apex engine : {} {}",
177 this.getClass().getName() + ":" + this.name, jmsMessage.getJMSMessageID(),
178 jmsMessage.getJMSType());
181 eventReceiver.receiveEvent(new Properties(), jmsMessage);
182 } catch (final Exception e) {
183 final String errorMessage = "failed to receive message from JMS";
184 LOGGER.warn(errorMessage, e);
185 throw new ApexEventRuntimeException(errorMessage, e);
194 stopOrderedFlag = true;
196 while (consumerThread.isAlive()) {
197 ThreadUtilities.sleep(jmsConsumerProperties.getConsumerWaitTime());
200 // Close the connection to the JMS server
202 if (connection != null) {
205 } catch (final Exception e) {
206 final String errorMessage = "close of connection to the JMS server failed";
207 LOGGER.warn(errorMessage, e);