gerrit.onap Code Review - policy/apex-pdp.git/blob

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2016-2018 Ericsson. All rights reserved.
   4  * ================================================================================
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  * SPDX-License-Identifier: Apache-2.0
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.policy.apex.testsuites.integration.uservice.adapt.jms;
  22
  23 import javax.jms.Connection;
  24 import javax.jms.ConnectionFactory;
  25 import javax.jms.JMSException;
  26 import javax.jms.Message;
  27 import javax.jms.MessageProducer;
  28 import javax.jms.Session;
  29 import javax.jms.Topic;
  30
  31 import org.apache.activemq.command.ActiveMQTopic;
  32 import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities;
  33 import org.onap.policy.apex.service.engine.event.ApexEventRuntimeException;
  34 import org.onap.policy.apex.testsuites.integration.common.testclasses.PingTestClass;
  35 import org.onap.policy.apex.testsuites.integration.uservice.adapt.events.EventGenerator;
  36 import org.slf4j.Logger;
  37 import org.slf4j.LoggerFactory;
  38
  39 /**
  40  * The Class JmsEventProducer.
  41  *
  42  * @author Liam Fallon (liam.fallon@ericsson.com)
  43  */
  44 public class JmsEventProducer implements Runnable {
  45     private static final Logger LOGGER = LoggerFactory.getLogger(JmsEventProducer.class);
  46
  47     private final String topic;
  48     private final int eventCount;
  49     private final boolean sendObjects;
  50     private final long eventInterval;
  51     private long eventsSentCount = 0;
  52
  53     private final Thread producerThread;
  54     private boolean sendEventsFlag = false;
  55     private boolean stopFlag = false;
  56     private final Connection connection;
  57
  58     /**
  59      * Instantiates a new jms event producer.
  60      *
  61      * @param topic the topic
  62      * @param connectionFactory the connection factory
  63      * @param username the username
  64      * @param password the password
  65      * @param eventCount the event count
  66      * @param sendObjects the send objects
  67      * @param eventInterval the event interval
  68      * @throws JMSException the JMS exception
  69      */
  70     public JmsEventProducer(final String topic, final ConnectionFactory connectionFactory, final String username,
  71             final String password, final int eventCount, final boolean sendObjects, final long eventInterval)
  72             throws JMSException {
  73         this.topic = topic;
  74         this.eventCount = eventCount;
  75         this.sendObjects = sendObjects;
  76         this.eventInterval = eventInterval;
  77         connection = connectionFactory.createConnection(username, password);
  78         connection.start();
  79
  80         producerThread = new Thread(this);
  81         producerThread.start();
  82     }
  83
  84     /**
  85      * {@inheritDoc}.
  86      */
  87     @Override
  88     public void run() {
  89         final Topic jmsTopic = new ActiveMQTopic(topic);
  90         try (final Session jmsSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  91                 final MessageProducer jmsProducer = jmsSession.createProducer(jmsTopic)) {
  92
  93             while (producerThread.isAlive() && !stopFlag) {
  94                 ThreadUtilities.sleep(50);
  95
  96                 if (sendEventsFlag) {
  97                     sendEventsToTopic(jmsSession, jmsProducer);
  98                     sendEventsFlag = false;
  99                 }
 100             }
 101
 102         } catch (final Exception e) {
 103             throw new ApexEventRuntimeException("JMS event consumption failed", e);
 104         }
 105     }
 106
 107     /**
 108      * Send events.
 109      */
 110     public void sendEvents() {
 111         sendEventsFlag = true;
 112     }
 113
 114     /**
 115      * Send events to topic.
 116      *
 117      * @param jmsSession the jms session
 118      * @param jmsProducer the jms producer
 119      * @throws JMSException the JMS exception
 120      */
 121     private void sendEventsToTopic(final Session jmsSession, final MessageProducer jmsProducer) throws JMSException {
 122
 123         LOGGER.info("{} : sending events to JMS server, event count {}", this.getClass().getCanonicalName(),
 124                 eventCount);
 125
 126         for (int i = 0; i < eventCount; i++) {
 127             ThreadUtilities.sleep(eventInterval);
 128
 129             Message jmsMessage = null;
 130             if (sendObjects) {
 131                 jmsMessage = jmsSession.createObjectMessage(new PingTestClass());
 132             } else {
 133                 jmsMessage = jmsSession.createTextMessage(EventGenerator.jsonEvent());
 134             }
 135             jmsProducer.send(jmsMessage);
 136             eventsSentCount++;
 137         }
 138         LOGGER.info("{} : completed, number of events sent", this.getClass().getCanonicalName(), eventsSentCount);
 139     }
 140
 141     /**
 142      * Gets the events sent count.
 143      *
 144      * @return the events sent count
 145      */
 146     public long getEventsSentCount() {
 147         return eventsSentCount;
 148     }
 149
 150     /**
 151      * Shutdown.
 152      */
 153     public void shutdown() {
 154         LOGGER.info("{} : stopping", this.getClass().getCanonicalName());
 155         stopFlag = true;
 156
 157         while (producerThread.isAlive()) {
 158             ThreadUtilities.sleep(10);
 159         }
 160         LOGGER.info("{} : stopped", this.getClass().getCanonicalName());
 161     }
 162
 163 }