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 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
   9  *
  10  *      http://www.apache.org/licenses/LICENSE-2.0
  11  *
  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.
  17  *
  18  * SPDX-License-Identifier: Apache-2.0
  19  * ============LICENSE_END=========================================================
  20  */
  21
  22 package org.onap.policy.apex.service.engine.main;
  23
  24 import java.util.List;
  25 import java.util.Properties;
  26 import java.util.concurrent.BlockingQueue;
  27 import java.util.concurrent.LinkedBlockingQueue;
  28 import java.util.concurrent.TimeUnit;
  29
  30 import org.onap.policy.apex.core.infrastructure.threading.ApplicationThreadFactory;
  31 import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities;
  32 import org.onap.policy.apex.model.basicmodel.concepts.ApexException;
  33 import org.onap.policy.apex.service.engine.event.ApexEvent;
  34 import org.onap.policy.apex.service.engine.event.ApexEventConsumer;
  35 import org.onap.policy.apex.service.engine.event.ApexEventException;
  36 import org.onap.policy.apex.service.engine.event.ApexEventProtocolConverter;
  37 import org.onap.policy.apex.service.engine.event.ApexEventReceiver;
  38 import org.onap.policy.apex.service.engine.event.PeeredReference;
  39 import org.onap.policy.apex.service.engine.event.SynchronousEventCache;
  40 import org.onap.policy.apex.service.engine.event.impl.EventConsumerFactory;
  41 import org.onap.policy.apex.service.engine.event.impl.EventProtocolFactory;
  42 import org.onap.policy.apex.service.parameters.engineservice.EngineServiceParameters;
  43 import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerParameters;
  44 import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerPeeredMode;
  45 import org.slf4j.ext.XLogger;
  46 import org.slf4j.ext.XLoggerFactory;
  47
  48 /**
  49  * This event unmarshaler handles events coming into Apex, handles threading, event queuing, transformation and
  50  * receiving using the configured receiving technology.
  51  *
  52  * @author Liam Fallon (liam.fallon@ericsson.com)
  53  */
  54 public class ApexEventUnmarshaller implements ApexEventReceiver, Runnable {
  55     // Get a reference to the logger
  56     private static final XLogger LOGGER = XLoggerFactory.getXLogger(ApexEventUnmarshaller.class);
  57
  58     // Interval to wait between thread shutdown checks
  59     private static final int UNMARSHALLER_SHUTDOWN_WAIT_INTERVAL = 10;
  60
  61     // The amount of time to wait between polls of the event queue in milliseconds
  62     private static final long EVENT_QUEUE_POLL_INTERVAL = 20;
  63
  64     // The name of the unmarshaler
  65     private final String name;
  66
  67     // The engine service and consumer parameters
  68     private final EngineServiceParameters engineServiceParameters;
  69     private final EventHandlerParameters consumerParameters;
  70
  71     // The engine service handler to use for forwarding on of unmarshalled events
  72     private ApexEngineServiceHandler engineServiceHandler;
  73
  74     // Apex event producer and event converter, all events are sent as string representations
  75     private ApexEventConsumer consumer;
  76     private ApexEventProtocolConverter converter;
  77
  78     // Temporary event holder for events going into Apex
  79     private final BlockingQueue<ApexEvent> queue = new LinkedBlockingQueue<>();
  80
  81     // The unmarshaler thread and stopping flag
  82     private Thread unmarshallerThread = null;
  83     private boolean stopOrderedFlag = false;
  84
  85     /**
  86      * Create the unmarshaler.
  87      *
  88      * @param name the name of the unmarshaler
  89      * @param engineServiceParameters the engine service parameters for this Apex engine
  90      * @param consumerParameters the consumer parameters for this specific unmarshaler
  91      */
  92     public ApexEventUnmarshaller(final String name, final EngineServiceParameters engineServiceParameters,
  93             final EventHandlerParameters consumerParameters) {
  94         this.name = name;
  95         this.engineServiceParameters = engineServiceParameters;
  96         this.consumerParameters = consumerParameters;
  97     }
  98
  99     /**
 100      * Configure the consumer and initialize the thread for event sending.
 101      *
 102      * @param incomingEngineServiceHandler the Apex engine service handler for passing events to Apex
 103      * @throws ApexEventException on errors initializing event handling
 104      */
 105     public void init(final ApexEngineServiceHandler incomingEngineServiceHandler) throws ApexEventException {
 106         this.engineServiceHandler = incomingEngineServiceHandler;
 107
 108         // Create the consumer for sending events and the converter for transforming events
 109         consumer = new EventConsumerFactory().createConsumer(name, consumerParameters);
 110         consumer.init(this.name, this.consumerParameters, this);
 111
 112         converter = new EventProtocolFactory().createConverter(name, consumerParameters.getEventProtocolParameters());
 113     }
 114
 115     /**
 116      * Start the unmarshaler and consumer threads.
 117      */
 118     public void start() {
 119         // Start the consumer
 120         consumer.start();
 121
 122         // Configure and start the event reception thread
 123         final String threadName =
 124                 engineServiceParameters.getEngineKey().getName() + ":" + this.getClass().getName() + ":" + name;
 125         unmarshallerThread = new ApplicationThreadFactory(threadName).newThread(this);
 126         unmarshallerThread.setDaemon(true);
 127         unmarshallerThread.start();
 128     }
 129
 130     /**
 131      * Gets the name of the unmarshaler.
 132      *
 133      * @return the unmarshaler name
 134      */
 135     public String getName() {
 136         return name;
 137     }
 138
 139     /**
 140      * Gets the technology specific consumer for this unmarshaler.
 141      *
 142      * @return the consumer
 143      */
 144     public ApexEventConsumer getConsumer() {
 145         return consumer;
 146     }
 147
 148     /**
 149      * Gets the event protocol converter for this unmarshaler.
 150      *
 151      * @return the event protocol converter
 152      */
 153     public ApexEventProtocolConverter getConverter() {
 154         return converter;
 155     }
 156
 157     /**
 158      * Connect a synchronous unmarshaler with a synchronous marshaler.
 159      *
 160      * @param peeredMode the peered mode under which the unmarshaler and marshaler are connected
 161      * @param peeredMarshaller the synchronous marshaler to connect with
 162      */
 163     public void connectMarshaler(final EventHandlerPeeredMode peeredMode, final ApexEventMarshaller peeredMarshaller) {
 164         switch (peeredMode) {
 165             case SYNCHRONOUS:
 166                 // To connect a synchronous unmarshaler and marshaler, we create a synchronous event
 167                 // cache on the consumer/producer pair
 168                 new SynchronousEventCache(peeredMode, consumer, peeredMarshaller.getProducer(),
 169                         consumerParameters.getPeerTimeout(EventHandlerPeeredMode.SYNCHRONOUS));
 170                 return;
 171
 172             case REQUESTOR:
 173                 new PeeredReference(peeredMode, consumer, peeredMarshaller.getProducer());
 174                 return;
 175
 176             default:
 177                 return;
 178         }
 179     }
 180
 181     /**
 182      * {@inheritDoc}.
 183      */
 184     @Override
 185     public void receiveEvent(final Properties executionProperties, final Object event) throws ApexEventException {
 186         receiveEvent(0, executionProperties, event, true);
 187     }
 188
 189     /**
 190      * {@inheritDoc}.
 191      */
 192     @Override
 193     public void receiveEvent(final long executionId, final Properties executionProperties, final Object event)
 194             throws ApexEventException {
 195         receiveEvent(executionId, executionProperties, event, false);
 196     }
 197
 198     /**
 199      * Receive an event from a consumer, convert its protocol and forward it to Apex.
 200      *
 201      * @param executionId the execution id the incoming execution ID
 202      * @param executionProperties properties used during processing of this event
 203      * @param event the event in its native format
 204      * @param generateExecutionId if true, let Apex generate the execution ID, if false, use the incoming execution ID
 205      * @throws ApexEventException on unmarshaling errors on events
 206      */
 207     private void receiveEvent(final long executionId, final Properties executionProperties, final Object event,
 208             final boolean generateExecutionId) throws ApexEventException {
 209         // Push the event onto the queue
 210         if (LOGGER.isTraceEnabled()) {
 211             String eventString = "onMessage(): event received: " + event.toString();
 212             LOGGER.trace(eventString);
 213         }
 214
 215         // Convert the incoming events to Apex events
 216         try {
 217             final List<ApexEvent> apexEventList = converter.toApexEvent(consumerParameters.getEventName(), event);
 218
 219             for (final ApexEvent apexEvent : apexEventList) {
 220                 isEventFilteredOut(apexEvent);
 221
 222                 // Check if this event is filtered out by the incoming filter
 223                 if (isEventFilteredOut(apexEvent)) {
 224                     // Ignore this event
 225                     continue;
 226                 }
 227
 228                 if (!generateExecutionId) {
 229                     apexEvent.setExecutionId(executionId);
 230                 }
 231
 232                 apexEvent.setExecutionProperties(executionProperties);
 233
 234                 // Enqueue the event
 235                 queue.add(apexEvent);
 236
 237                 // Cache synchronized events that are sent
 238                 if (consumerParameters.isPeeredMode(EventHandlerPeeredMode.SYNCHRONOUS)) {
 239                     final SynchronousEventCache synchronousEventCache =
 240                             (SynchronousEventCache) consumer.getPeeredReference(EventHandlerPeeredMode.SYNCHRONOUS);
 241                     synchronousEventCache.cacheSynchronizedEventToApex(apexEvent.getExecutionId(), apexEvent);
 242                 }
 243             }
 244         } catch (final ApexException e) {
 245             final String errorMessage = "Error while converting event into an ApexEvent for " + name + ": "
 246                     + e.getMessage() + ", Event=" + event;
 247             LOGGER.warn(errorMessage, e);
 248             throw new ApexEventException(errorMessage, e);
 249         }
 250     }
 251
 252     /**
 253      * Check if an event is filtered out and ignored.
 254      *
 255      * @param apexEvent the event to check
 256      */
 257     private boolean isEventFilteredOut(final ApexEvent apexEvent) {
 258         // Check if we are filtering events on this unmarshaler, if so check the event name
 259         // against the filter
 260         if (consumerParameters.isSetEventNameFilter()
 261                 && !apexEvent.getName().matches(consumerParameters.getEventNameFilter())) {
 262
 263             if (LOGGER.isTraceEnabled()) {
 264                 LOGGER.trace("onMessage(): event {} not processed, filtered  out by filter", apexEvent,
 265                         consumerParameters.getEventNameFilter());
 266             }
 267
 268             return true;
 269         } else {
 270             return false;
 271         }
 272     }
 273
 274     /**
 275      * Run a thread that runs forever (well until system termination anyway) and listens for incoming events on the
 276      * queue.
 277      */
 278     @Override
 279     public void run() {
 280         // Run until interruption
 281         while (unmarshallerThread.isAlive() && !stopOrderedFlag) {
 282             try {
 283                 // Take the next event from the queue
 284                 final ApexEvent apexEvent = queue.poll(EVENT_QUEUE_POLL_INTERVAL, TimeUnit.MILLISECONDS);
 285                 if (apexEvent == null) {
 286                     continue;
 287                 }
 288
 289                 if (LOGGER.isTraceEnabled()) {
 290                     String message = apexEvent.toString();
 291                     LOGGER.trace("event received {}", message);
 292                 }
 293
 294                 // Pass the event to the activator for forwarding to Apex
 295                 engineServiceHandler.forwardEvent(apexEvent);
 296             } catch (final InterruptedException e) {
 297                 // restore the interrupt status
 298                 Thread.currentThread().interrupt();
 299                 LOGGER.warn("BatchProcessor thread interrupted, Reason {}", e.getMessage());
 300                 stopOrderedFlag = true;
 301             } catch (final Exception e) {
 302                 LOGGER.warn("Error while forwarding events for " + unmarshallerThread.getName(), e);
 303             }
 304         }
 305
 306         // Stop event production
 307         consumer.stop();
 308     }
 309
 310     /**
 311      * Get the unmarshaler thread.
 312      *
 313      * @return the unmarshaler thread
 314      */
 315     public Thread getThread() {
 316         return unmarshallerThread;
 317     }
 318
 319     /**
 320      * Stop the Apex event unmarshaller's event producer using its termination mechanism.
 321      */
 322     public void stop() {
 323         LOGGER.entry("shutting down Apex event unmarshaller . . .");
 324
 325         // Order the stop
 326         stopOrderedFlag = true;
 327
 328         // Order a stop on the synchronous cache if one exists
 329         if (consumerParameters != null && consumerParameters.isPeeredMode(EventHandlerPeeredMode.SYNCHRONOUS)
 330                 && consumer.getPeeredReference(EventHandlerPeeredMode.SYNCHRONOUS) != null) {
 331             ((SynchronousEventCache) consumer.getPeeredReference(EventHandlerPeeredMode.SYNCHRONOUS)).stop();
 332         }
 333
 334         // Wait for thread shutdown
 335         while (unmarshallerThread != null && unmarshallerThread.isAlive()) {
 336             ThreadUtilities.sleep(UNMARSHALLER_SHUTDOWN_WAIT_INTERVAL);
 337         }
 338
 339         LOGGER.exit("shut down Apex event unmarshaller");
 340     }
 341 }