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.service.engine.main;
  22
  23 import java.util.concurrent.BlockingQueue;
  24 import java.util.concurrent.LinkedBlockingQueue;
  25 import java.util.concurrent.TimeUnit;
  26
  27 import org.onap.policy.apex.core.infrastructure.threading.ApplicationThreadFactory;
  28 import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities;
  29 import org.onap.policy.apex.service.engine.event.ApexEvent;
  30 import org.onap.policy.apex.service.engine.event.ApexEventException;
  31 import org.onap.policy.apex.service.engine.event.ApexEventProducer;
  32 import org.onap.policy.apex.service.engine.event.ApexEventProtocolConverter;
  33 import org.onap.policy.apex.service.engine.event.impl.EventProducerFactory;
  34 import org.onap.policy.apex.service.engine.event.impl.EventProtocolFactory;
  35 import org.onap.policy.apex.service.engine.runtime.ApexEventListener;
  36 import org.onap.policy.apex.service.parameters.engineservice.EngineServiceParameters;
  37 import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerParameters;
  38 import org.slf4j.ext.XLogger;
  39 import org.slf4j.ext.XLoggerFactory;
  40
  41 /**
  42  * This event marshaler handles events coming out of Apex and sends them on, handles threading,
  43  * event queuing, transformations and sending using the configured sending technology.
  44  *
  45  * @author Liam Fallon (liam.fallon@ericsson.com)
  46  */
  47 public class ApexEventMarshaller implements ApexEventListener, Runnable {
  48     // Get a reference to the logger
  49     private static final XLogger LOGGER = XLoggerFactory.getXLogger(ApexEventMarshaller.class);
  50
  51     // Interval to wait between thread shutdown checks
  52     private static final int MARSHALLER_SHUTDOWN_WAIT_INTERVAL = 10;
  53
  54     // The amount of time to wait between polls of the event queue in milliseconds
  55     private static final long EVENT_QUEUE_POLL_INTERVAL = 20;
  56
  57     // The name of the marshaler
  58     private final String name;
  59
  60     // The engine service and producer parameters
  61     private final EngineServiceParameters engineServiceParameters;
  62     private final EventHandlerParameters producerParameters;
  63
  64     // Apex event producer and event converter, all conversions are to and from string
  65     // representation of events
  66     private ApexEventProducer producer;
  67     private ApexEventProtocolConverter converter;
  68
  69     // Temporary event holder for events coming out of Apex
  70     private final BlockingQueue<ApexEvent> queue = new LinkedBlockingQueue<>();
  71
  72     // The marshaler thread and stopping flag
  73     private Thread marshallerThread;
  74     private boolean stopOrderedFlag = false;
  75
  76     /**
  77      * Create the marshaler.
  78      *
  79      * @param name the name of the marshaler
  80      * @param engineServiceParameters the engine service parameters for this Apex engine
  81      * @param producerParameters the producer parameters for this specific marshaler
  82      */
  83     public ApexEventMarshaller(final String name, final EngineServiceParameters engineServiceParameters,
  84             final EventHandlerParameters producerParameters) {
  85         this.name = name;
  86         this.engineServiceParameters = engineServiceParameters;
  87         this.producerParameters = producerParameters;
  88     }
  89
  90     /**
  91      * Configure the marshaler by setting up the producer and event converter and initialize the
  92      * thread for event sending.
  93      *
  94      * @throws ApexActivatorException on errors initializing the producer
  95      * @throws ApexEventException on errors initializing event handling
  96      */
  97     public void init() throws ApexActivatorException, ApexEventException {
  98         // Create the producer for sending events and the converter for transforming events
  99         producer = new EventProducerFactory().createProducer(name, producerParameters);
 100
 101         // Initialize the producer
 102         producer.init(this.name, this.producerParameters);
 103
 104         // Create the converter for transforming events
 105         converter = new EventProtocolFactory().createConverter(name, producerParameters.getEventProtocolParameters());
 106
 107         // Configure and start the event sending thread
 108         final String threadName =
 109                 engineServiceParameters.getEngineKey().getName() + ':' + this.getClass().getName() + ':' + this.name;
 110         marshallerThread = new ApplicationThreadFactory(threadName).newThread(this);
 111         marshallerThread.setDaemon(true);
 112         marshallerThread.start();
 113     }
 114
 115     /**
 116      * Gets the name of the marshaler.
 117      *
 118      * @return the marshaler name
 119      */
 120     public String getName() {
 121         return name;
 122     }
 123
 124     /**
 125      * Gets the technology specific producer for this marshaler.
 126      *
 127      * @return the producer
 128      */
 129     public ApexEventProducer getProducer() {
 130         return producer;
 131     }
 132
 133     /**
 134      * Gets the event protocol converter for this marshaler.
 135      *
 136      * @return the event protocol converter
 137      */
 138     public ApexEventProtocolConverter getConverter() {
 139         return converter;
 140     }
 141
 142     /**
 143      * Callback method called on implementations of this interface when Apex emits an event.
 144      *
 145      * @param apexEvent the apex event emitted by Apex
 146      */
 147     @Override
 148     public void onApexEvent(final ApexEvent apexEvent) {
 149         // Check if we are filtering events on this marshaler, if so check the event name against
 150         // the filter
 151         if (producerParameters.isSetEventNameFilter()
 152                 && !apexEvent.getName().matches(producerParameters.getEventNameFilter())) {
 153             if (LOGGER.isTraceEnabled()) {
 154                 LOGGER.trace("onMessage(): event {} not processed, filtered  out by filter", apexEvent,
 155                         producerParameters.getEventNameFilter());
 156             }
 157
 158             // Ignore this event
 159             return;
 160         }
 161
 162         // Push the event onto the queue for handling
 163         try {
 164             queue.put(apexEvent);
 165         } catch (final InterruptedException e) {
 166             // restore the interrupt status
 167             Thread.currentThread().interrupt();
 168             LOGGER.warn("Failed to queue the event: " + apexEvent, e);
 169         }
 170     }
 171
 172     /**
 173      * Run a thread that runs forever (well until system termination anyway) and listens for
 174      * outgoing events on the queue.
 175      */
 176     @Override
 177     public void run() {
 178         // Run until interrupted
 179         while (marshallerThread.isAlive() && !stopOrderedFlag) {
 180             try {
 181                 // Take the next event from the queue
 182                 final ApexEvent apexEvent = queue.poll(EVENT_QUEUE_POLL_INTERVAL, TimeUnit.MILLISECONDS);
 183                 if (apexEvent == null) {
 184                     continue;
 185                 }
 186
 187                 // Process the next Apex event from the queue
 188                 final Object event = converter.fromApexEvent(apexEvent);
 189
 190                 producer.sendEvent(apexEvent.getExecutionId(), apexEvent.getName(), event);
 191
 192                 if (LOGGER.isTraceEnabled()) {
 193                     String message = "event sent : " + apexEvent.toString();
 194                     LOGGER.trace(message);
 195                 }
 196             } catch (final InterruptedException e) {
 197                 // restore the interrupt status
 198                 Thread.currentThread().interrupt();
 199                 LOGGER.debug("Thread interrupted, Reason {}", e.getMessage());
 200                 stopOrderedFlag = true;
 201             } catch (final Exception e) {
 202                 LOGGER.warn("Error while forwarding events for " + marshallerThread.getName(), e);
 203             }
 204         }
 205
 206         // Stop event production if we are not synchronized,;in the synchronized case, the producer
 207         // takes care of its own cleanup.
 208         producer.stop();
 209     }
 210
 211     /**
 212      * Get the marshaler thread.
 213      *
 214      * @return the marshaler thread
 215      */
 216     public Thread getThread() {
 217         return marshallerThread;
 218     }
 219
 220     /**
 221      * Stop the Apex event marshaller's event producer using its termination mechanism.
 222      */
 223     public void stop() {
 224         LOGGER.entry("shutting down Apex event marshaller . . .");
 225
 226         // Order the stop
 227         stopOrderedFlag = true;
 228
 229         // Wait for thread shutdown
 230         while (marshallerThread.isAlive()) {
 231             ThreadUtilities.sleep(MARSHALLER_SHUTDOWN_WAIT_INTERVAL);
 232         }
 233
 234         LOGGER.exit("shut down Apex event marshaller");
 235     }
 236 }