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

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2016-2018 Ericsson. All rights reserved.
   4  *  Modifications Copyright (C) 2019-2020 Nordix Foundation.
   5  *  Modifications Copyright (C) 2021 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  *
  19  * SPDX-License-Identifier: Apache-2.0
  20  * ============LICENSE_END=========================================================
  21  */
  22
  23 package org.onap.policy.apex.testsuites.integration.uservice.adapt.kafka;
  24
  25 import com.salesforce.kafka.test.junit4.SharedKafkaTestResource;
  26 import java.time.Duration;
  27 import lombok.Getter;
  28 import org.apache.kafka.clients.producer.Producer;
  29 import org.apache.kafka.clients.producer.ProducerRecord;
  30 import org.apache.kafka.common.serialization.StringSerializer;
  31 import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities;
  32 import org.onap.policy.apex.testsuites.integration.uservice.adapt.events.EventGenerator;
  33 import org.slf4j.Logger;
  34 import org.slf4j.LoggerFactory;
  35
  36 /**
  37  * The Class KafkaEventProducer.
  38  *
  39  * @author Liam Fallon (liam.fallon@ericsson.com)
  40  */
  41 public class KafkaEventProducer implements Runnable {
  42     // Get a reference to the logger
  43     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaEventProducer.class);
  44
  45     private final String topic;
  46     private final SharedKafkaTestResource sharedKafkaTestResource;
  47     private final int eventCount;
  48     private final boolean xmlEvents;
  49     private final long eventInterval;
  50     @Getter
  51     private long eventsSentCount = 0;
  52
  53     private final Thread producerThread;
  54     private boolean sendEventsFlag = false;
  55     private volatile boolean stopFlag = false;
  56
  57     /**
  58      * Instantiates a new kafka event producer.
  59      *
  60      * @param topic the topic
  61      * @param sharedKafkaTestResource the kafka server address
  62      * @param eventCount the event count
  63      * @param xmlEvents the xml events
  64      * @param eventInterval the event interval
  65      */
  66     public KafkaEventProducer(final String topic, final SharedKafkaTestResource sharedKafkaTestResource,
  67         final int eventCount, final boolean xmlEvents, final long eventInterval) {
  68         this.topic = topic;
  69         this.sharedKafkaTestResource = sharedKafkaTestResource;
  70         this.eventCount = eventCount;
  71         this.xmlEvents = xmlEvents;
  72         this.eventInterval = eventInterval;
  73
  74         producerThread = new Thread(this);
  75         producerThread.start();
  76     }
  77
  78     /**
  79      * {@inheritDoc}.
  80      */
  81     @Override
  82     public void run() {
  83         final Producer<String, String> producer = sharedKafkaTestResource.getKafkaTestUtils()
  84             .getKafkaProducer(StringSerializer.class, StringSerializer.class);
  85
  86         while (producerThread.isAlive() && !stopFlag) {
  87             ThreadUtilities.sleep(50);
  88
  89             if (sendEventsFlag) {
  90                 sendEventsToTopic(producer);
  91                 sendEventsFlag = false;
  92             }
  93         }
  94
  95         producer.close(Duration.ofMillis(1000));
  96     }
  97
  98     /**
  99      * Send events.
 100      */
 101     public void sendEvents() {
 102         sendEventsFlag = true;
 103     }
 104
 105     /**
 106      * Send events to topic.
 107      *
 108      * @param producer the producer
 109      */
 110     private void sendEventsToTopic(final Producer<String, String> producer) {
 111         LOGGER.debug("{} : sending events to Kafka server, event count {}, xmlEvents {}",
 112             KafkaEventProducer.class.getName(), eventCount, xmlEvents);
 113
 114         for (int i = 0; i < eventCount; i++) {
 115             LOGGER.debug("{} : waiting {} milliseconds before sending next event", KafkaEventProducer.class.getName(),
 116                 eventInterval);
 117             ThreadUtilities.sleep(eventInterval);
 118
 119             String eventString = null;
 120             if (xmlEvents) {
 121                 eventString = EventGenerator.xmlEvent();
 122             } else {
 123                 eventString = EventGenerator.jsonEvent();
 124             }
 125             producer.send(new ProducerRecord<String, String>(topic, "Event" + i + "Of" + eventCount, eventString));
 126             producer.flush();
 127             eventsSentCount++;
 128             LOGGER.debug("****** Sent event No. {} ******\n{}", eventsSentCount, eventString);
 129         }
 130         LOGGER.debug("{}: completed", KafkaEventProducer.class.getName());
 131     }
 132
 133     /**
 134      * Shutdown.
 135      */
 136     public void shutdown() {
 137         LOGGER.debug("{} : stopping", KafkaEventProducer.class.getName());
 138
 139         stopFlag = true;
 140
 141         while (producerThread.isAlive()) {
 142             ThreadUtilities.sleep(10);
 143         }
 144
 145         LOGGER.debug("{} : stopped", KafkaEventProducer.class.getName());
 146     }
 147
 148     public boolean isAlive() {
 149         return producerThread.isAlive();
 150     }
 151 }