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
9 * http://www.apache.org/licenses/LICENSE-2.0
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.
17 * SPDX-License-Identifier: Apache-2.0
18 * ============LICENSE_END=========================================================
21 package org.onap.policy.apex.service.engine.main;
23 import java.util.concurrent.BlockingQueue;
24 import java.util.concurrent.LinkedBlockingQueue;
25 import java.util.concurrent.TimeUnit;
26 import org.onap.policy.apex.core.infrastructure.threading.ApplicationThreadFactory;
27 import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities;
28 import org.onap.policy.apex.service.engine.event.ApexEvent;
29 import org.onap.policy.apex.service.engine.event.ApexEventException;
30 import org.onap.policy.apex.service.engine.event.ApexEventProducer;
31 import org.onap.policy.apex.service.engine.event.ApexEventProtocolConverter;
32 import org.onap.policy.apex.service.engine.event.impl.EventProducerFactory;
33 import org.onap.policy.apex.service.engine.event.impl.EventProtocolFactory;
34 import org.onap.policy.apex.service.engine.runtime.ApexEventListener;
35 import org.onap.policy.apex.service.parameters.engineservice.EngineServiceParameters;
36 import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerParameters;
37 import org.slf4j.ext.XLogger;
38 import org.slf4j.ext.XLoggerFactory;
41 * This event marshaler handles events coming out of Apex and sends them on, handles threading, event queuing,
42 * transformations and sending using the configured sending technology.
44 * @author Liam Fallon (liam.fallon@ericsson.com)
46 public class ApexEventMarshaller implements ApexEventListener, Runnable {
47 // Get a reference to the logger
48 private static final XLogger LOGGER = XLoggerFactory.getXLogger(ApexEventMarshaller.class);
50 // Interval to wait between thread shutdown checks
51 private static final int MARSHALLER_SHUTDOWN_WAIT_INTERVAL = 10;
53 // The amount of time to wait between polls of the event queue in milliseconds
54 private static final long EVENT_QUEUE_POLL_INTERVAL = 20;
56 // The name of the marshaler
57 private final String name;
59 // The engine service and producer parameters
60 private final EngineServiceParameters engineServiceParameters;
61 private final EventHandlerParameters producerParameters;
63 // Apex event producer and event converter, all conversions are to and from string
64 // representation of events
65 private ApexEventProducer producer;
66 private ApexEventProtocolConverter converter;
68 // Temporary event holder for events coming out of Apex
69 private final BlockingQueue<ApexEvent> queue = new LinkedBlockingQueue<>();
71 // The marshaler thread and stopping flag
72 private Thread marshallerThread;
73 private boolean stopOrderedFlag = false;
76 * Create the marshaler.
78 * @param name the name of the marshaler
79 * @param engineServiceParameters the engine service parameters for this Apex engine
80 * @param producerParameters the producer parameters for this specific marshaler
82 public ApexEventMarshaller(final String name, final EngineServiceParameters engineServiceParameters,
83 final EventHandlerParameters producerParameters) {
85 this.engineServiceParameters = engineServiceParameters;
86 this.producerParameters = producerParameters;
90 * Configure the marshaler by setting up the producer and event converter and initialize the thread for event
93 * @throws ApexEventException on errors initializing event handling
95 public void init() throws ApexEventException {
96 // Create the producer for sending events and the converter for transforming events
97 producer = new EventProducerFactory().createProducer(name, producerParameters);
99 // Initialize the producer
100 producer.init(this.name, this.producerParameters);
102 // Create the converter for transforming events
103 converter = new EventProtocolFactory().createConverter(name, producerParameters.getEventProtocolParameters());
105 // Configure and start the event sending thread
106 final String threadName =
107 engineServiceParameters.getEngineKey().getName() + ':' + this.getClass().getName() + ':' + this.name;
108 marshallerThread = new ApplicationThreadFactory(threadName).newThread(this);
109 marshallerThread.setDaemon(true);
110 marshallerThread.start();
114 * Gets the name of the marshaler.
116 * @return the marshaler name
118 public String getName() {
123 * Gets the technology specific producer for this marshaler.
125 * @return the producer
127 public ApexEventProducer getProducer() {
132 * Gets the event protocol converter for this marshaler.
134 * @return the event protocol converter
136 public ApexEventProtocolConverter getConverter() {
141 * Callback method called on implementations of this interface when Apex emits an event.
143 * @param apexEvent the apex event emitted by Apex
146 public void onApexEvent(final ApexEvent apexEvent) {
147 // Check if we are filtering events on this marshaler, if so check the event name against
149 if (producerParameters.isSetEventNameFilter()
150 && !apexEvent.getName().matches(producerParameters.getEventNameFilter())) {
151 if (LOGGER.isTraceEnabled()) {
152 LOGGER.trace("onMessage(): event {} not processed, filtered out by filter", apexEvent,
153 producerParameters.getEventNameFilter());
160 // Push the event onto the queue for handling
162 queue.put(apexEvent);
163 } catch (final InterruptedException e) {
164 // restore the interrupt status
165 Thread.currentThread().interrupt();
166 LOGGER.warn("Queueing of the event was interrupted: " + apexEvent, e);
171 * Run a thread that runs forever (well until system termination anyway) and listens for outgoing events on the
176 // Run until interrupted
177 while (marshallerThread.isAlive() && !stopOrderedFlag) {
179 // Take the next event from the queue
180 final ApexEvent apexEvent = queue.poll(EVENT_QUEUE_POLL_INTERVAL, TimeUnit.MILLISECONDS);
181 if (apexEvent == null) {
185 // Process the next Apex event from the queue
186 final Object event = converter.fromApexEvent(apexEvent);
188 producer.sendEvent(apexEvent.getExecutionId(), apexEvent.getExecutionProperties(), apexEvent.getName(),
191 if (LOGGER.isTraceEnabled()) {
192 final String message = "event sent : " + apexEvent.toString();
193 LOGGER.trace(message);
195 } catch (final InterruptedException e) {
196 // restore the interrupt status
197 Thread.currentThread().interrupt();
198 LOGGER.debug("Thread interrupted, Reason {}", e.getMessage());
199 stopOrderedFlag = true;
200 } catch (final Exception e) {
201 LOGGER.warn("Error while forwarding events for " + marshallerThread.getName(), e);
205 // Stop event production if we are not synchronized,;in the synchronized case, the producer
206 // takes care of its own cleanup.
211 * Get the marshaler thread.
213 * @return the marshaler thread
215 public Thread getThread() {
216 return marshallerThread;
220 * Stop the Apex event marshaller's event producer using its termination mechanism.
223 LOGGER.entry("shutting down Apex event marshaller . . .");
226 stopOrderedFlag = true;
228 // Wait for thread shutdown
229 while (marshallerThread.isAlive()) {
230 ThreadUtilities.sleep(MARSHALLER_SHUTDOWN_WAIT_INTERVAL);
233 LOGGER.exit("shut down Apex event marshaller");