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.runtime.impl;
  22
  23 import com.google.gson.Gson;
  24 import com.google.gson.GsonBuilder;
  25 import com.google.gson.JsonElement;
  26 import com.google.gson.JsonParser;
  27
  28 import java.io.ByteArrayInputStream;
  29 import java.io.ByteArrayOutputStream;
  30 import java.util.Arrays;
  31 import java.util.Collection;
  32 import java.util.Map;
  33 import java.util.Map.Entry;
  34 import java.util.concurrent.BlockingQueue;
  35
  36 import org.onap.policy.apex.context.ContextException;
  37 import org.onap.policy.apex.context.ContextRuntimeException;
  38 import org.onap.policy.apex.context.SchemaHelper;
  39 import org.onap.policy.apex.context.impl.schema.SchemaHelperFactory;
  40 import org.onap.policy.apex.core.engine.engine.ApexEngine;
  41 import org.onap.policy.apex.core.engine.engine.impl.ApexEngineFactory;
  42 import org.onap.policy.apex.core.engine.event.EnEvent;
  43 import org.onap.policy.apex.core.infrastructure.threading.ApplicationThreadFactory;
  44 import org.onap.policy.apex.model.basicmodel.concepts.ApexException;
  45 import org.onap.policy.apex.model.basicmodel.concepts.AxArtifactKey;
  46 import org.onap.policy.apex.model.basicmodel.handling.ApexModelException;
  47 import org.onap.policy.apex.model.basicmodel.handling.ApexModelReader;
  48 import org.onap.policy.apex.model.basicmodel.handling.ApexModelWriter;
  49 import org.onap.policy.apex.model.basicmodel.service.ModelService;
  50 import org.onap.policy.apex.model.contextmodel.concepts.AxContextAlbum;
  51 import org.onap.policy.apex.model.contextmodel.concepts.AxContextAlbums;
  52 import org.onap.policy.apex.model.enginemodel.concepts.AxEngineModel;
  53 import org.onap.policy.apex.model.enginemodel.concepts.AxEngineState;
  54 import org.onap.policy.apex.model.policymodel.concepts.AxPolicyModel;
  55 import org.onap.policy.apex.service.engine.event.ApexEvent;
  56 import org.onap.policy.apex.service.engine.event.impl.enevent.ApexEvent2EnEventConverter;
  57 import org.onap.policy.apex.service.engine.runtime.ApexEventListener;
  58 import org.onap.policy.apex.service.engine.runtime.EngineService;
  59 import org.onap.policy.apex.service.engine.runtime.EngineServiceEventInterface;
  60 import org.slf4j.ext.XLogger;
  61 import org.slf4j.ext.XLoggerFactory;
  62
  63 /**
  64  * The Class EngineWorker encapsulates a core {@link ApexEngine} instance, which runs policies
  65  * defined in the {@link org.onap.policy.apex.model.basicmodel.concepts.AxModelAxModel}. Each policy
  66  * is triggered by an Apex event, and when the policy is triggered it runs through to completion in
  67  * the ApexEngine.
  68  *
  69  * <p>This class acts as a container for an {@link ApexEngine}, running it in a thread, sending it
  70  * events, and receiving events from it.
  71  *
  72  * @author Liam Fallon (liam.fallon@ericsson.com)
  73  */
  74 final class EngineWorker implements EngineService {
  75     // Logger for this class
  76     private static final XLogger LOGGER = XLoggerFactory.getXLogger(EngineService.class);
  77
  78     // Recurring string constants
  79     private static final String IS_NULL_SUFFIX = " is  null";
  80     private static final String ENGINE_FOR_KEY_PREFIX = "apex engine for engine key ";
  81     private static final String ENGINE_SUFFIX = " of this engine";
  82     private static final String BAD_KEY_MATCH_TAG = " does not match the key";
  83     private static final String ENGINE_KEY_PREFIX = "engine key ";
  84
  85     // The ID of this engine
  86     private final AxArtifactKey engineWorkerKey;
  87
  88     // The Apex engine which is running the policies in this worker
  89     private final ApexEngine engine;
  90
  91     // The event processor is an inner class, an instance of which runs as a thread that reads
  92     // incoming events from a queue and forwards them to the Apex engine
  93     private EventProcessor processor = null;
  94
  95     // Thread handling for the worker
  96     private final ApplicationThreadFactory threadFactory;
  97     private Thread processorThread;
  98
  99     // Converts ApexEvent instances to and from EnEvent instances
 100     private ApexEvent2EnEventConverter apexEnEventConverter = null;
 101
 102     /**
 103      * Constructor that creates an Apex engine, an event processor for events to be sent to that
 104      * engine, and an {@link ApexModelReader} instance to read Apex models using JAXB.
 105      *
 106      * @param engineWorkerKey the engine worker key
 107      * @param queue the queue on which events for this Apex worker will come
 108      * @param threadFactory the thread factory to use for creating the event processing thread
 109      * @throws ApexException thrown on errors on worker instantiation
 110      */
 111     EngineWorker(final AxArtifactKey engineWorkerKey, final BlockingQueue<ApexEvent> queue,
 112             final ApplicationThreadFactory threadFactory) {
 113         LOGGER.entry(engineWorkerKey);
 114
 115         this.engineWorkerKey = engineWorkerKey;
 116         this.threadFactory = threadFactory;
 117
 118         // Create the Apex engine
 119         engine = new ApexEngineFactory().createApexEngine(engineWorkerKey);
 120
 121         // Create and run the event processor
 122         processor = new EventProcessor(queue);
 123
 124         // Set the Event converter up
 125         apexEnEventConverter = new ApexEvent2EnEventConverter(engine);
 126
 127         LOGGER.exit();
 128     }
 129
 130     /*
 131      * (non-Javadoc)
 132      *
 133      * @see
 134      * org.onap.policy.apex.service.engine.runtime.EngineService#registerActionListener(java.lang.
 135      * String, org.onap.policy.apex.service.engine.runtime.ApexEventListener)
 136      */
 137     @Override
 138     public void registerActionListener(final String listenerName, final ApexEventListener apexEventListener) {
 139         // Sanity checks on the Apex model
 140         if (engine == null) {
 141             LOGGER.warn("listener registration on engine with key " + engineWorkerKey.getId()
 142                     + ", failed, listener is null");
 143             return;
 144         }
 145
 146         engine.addEventListener(listenerName, new EnEventListenerImpl(apexEventListener, apexEnEventConverter));
 147     }
 148
 149     /*
 150      * (non-Javadoc)
 151      *
 152      * @see
 153      * org.onap.policy.apex.service.engine.runtime.EngineService#deregisterActionListener(java.lang.
 154      * String)
 155      */
 156     @Override
 157     public void deregisterActionListener(final String listenerName) {
 158         // Sanity checks on the Apex model
 159         if (engine == null) {
 160             LOGGER.warn("listener deregistration on engine with key " + engineWorkerKey.getId()
 161                     + ", failed, listener is null");
 162             return;
 163         }
 164
 165         engine.removeEventListener(listenerName);
 166     }
 167
 168     /*
 169      * (non-Javadoc)
 170      *
 171      * @see
 172      * org.onap.policy.apex.service.engine.runtime.EngineService#getEngineServiceEventInterface()
 173      */
 174     @Override
 175     public EngineServiceEventInterface getEngineServiceEventInterface() {
 176         throw new UnsupportedOperationException(
 177                 "getEngineServiceEventInterface() call is not allowed on an Apex Engine Worker");
 178     }
 179
 180     /*
 181      * (non-Javadoc)
 182      *
 183      * @see org.onap.policy.apex.service.engine.runtime.EngineService#getKey()
 184      */
 185     @Override
 186     public AxArtifactKey getKey() {
 187         return engineWorkerKey;
 188     }
 189
 190     /*
 191      * (non-Javadoc)
 192      *
 193      * @see org.onap.policy.apex.service.engine.runtime.EngineService#getInfo()
 194      */
 195     @Override
 196     public Collection<AxArtifactKey> getEngineKeys() {
 197         return Arrays.asList(engineWorkerKey);
 198     }
 199
 200     /*
 201      * (non-Javadoc)
 202      *
 203      * @see org.onap.policy.apex.service.engine.runtime.EngineService#getApexModelKey()
 204      */
 205     @Override
 206     public AxArtifactKey getApexModelKey() {
 207         if (ModelService.existsModel(AxPolicyModel.class)) {
 208             return ModelService.getModel(AxPolicyModel.class).getKey();
 209         } else {
 210             return null;
 211         }
 212     }
 213
 214     /*
 215      * (non-Javadoc)
 216      *
 217      * @see
 218      * org.onap.policy.apex.service.engine.runtime.EngineService#updateModel(org.onap.policy.apex.
 219      * model. basicmodel.concepts.AxArtifactKey, java.lang.String, boolean)
 220      */
 221     @Override
 222     public void updateModel(final AxArtifactKey engineKey, final String engineModel, final boolean forceFlag)
 223             throws ApexException {
 224         LOGGER.entry(engineKey);
 225
 226         // Read the Apex model into memory using the Apex Model Reader
 227         AxPolicyModel apexPolicyModel = null;
 228         try {
 229             final ApexModelReader<AxPolicyModel> modelReader = new ApexModelReader<>(AxPolicyModel.class);
 230             apexPolicyModel = modelReader.read(new ByteArrayInputStream(engineModel.getBytes()));
 231         } catch (final ApexModelException e) {
 232             LOGGER.error("failed to unmarshal the apex model on engine " + engineKey.getId(), e);
 233             throw new ApexException("failed to unmarshal the apex model on engine " + engineKey.getId(), e);
 234         }
 235
 236         if (apexPolicyModel == null) {
 237             LOGGER.error("apex model null on engine " + engineKey.getId());
 238             throw new ApexException("apex model null on engine " + engineKey.getId());
 239         }
 240
 241         // Update the Apex model in the Apex engine
 242         updateModel(engineKey, apexPolicyModel, forceFlag);
 243
 244         LOGGER.exit();
 245     }
 246
 247     /*
 248      * (non-Javadoc)
 249      *
 250      * @see
 251      * org.onap.policy.apex.service.engine.runtime.EngineService#updateModel(org.onap.policy.apex.
 252      * model. basicmodel.concepts.AxArtifactKey,
 253      * org.onap.policy.apex.model.policymodel.concepts.AxPolicyModel, boolean)
 254      */
 255     @Override
 256     public void updateModel(final AxArtifactKey engineKey, final AxPolicyModel apexModel, final boolean forceFlag)
 257             throws ApexException {
 258         LOGGER.entry(engineKey);
 259
 260         // Check if the key on the update request is correct
 261         if (!engineWorkerKey.equals(engineKey)) {
 262             String message = ENGINE_KEY_PREFIX + engineKey.getId() + BAD_KEY_MATCH_TAG + engineWorkerKey.getId()
 263                     + ENGINE_SUFFIX;
 264             LOGGER.warn(message);
 265             throw new ApexException(message);
 266         }
 267
 268         // Sanity checks on the Apex model
 269         if (engine == null) {
 270             LOGGER.warn("engine with key " + engineKey.getId() + " not initialized");
 271             throw new ApexException("engine with key " + engineKey.getId() + " not initialized");
 272         }
 273
 274         // Check model compatibility
 275         if (ModelService.existsModel(AxPolicyModel.class)) {
 276             // The current policy model may or may not be defined
 277             final AxPolicyModel currentModel = ModelService.getModel(AxPolicyModel.class);
 278             if (!currentModel.getKey().isCompatible(apexModel.getKey())) {
 279                 if (forceFlag) {
 280                     LOGGER.warn("apex model update forced, supplied model with key \"" + apexModel.getKey().getId()
 281                             + "\" is not a compatible model update from the existing engine model with key \""
 282                             + currentModel.getKey().getId() + "\"");
 283                 } else {
 284                     throw new ContextException(
 285                             "apex model update failed, supplied model with key \"" + apexModel.getKey().getId()
 286                                     + "\" is not a compatible model update from the existing engine model with key \""
 287                                     + currentModel.getKey().getId() + "\"");
 288                 }
 289             }
 290         }
 291
 292         // Update the Apex model in the Apex engine
 293         engine.clear();
 294         engine.updateModel(apexModel);
 295
 296         LOGGER.debug("engine model {} added to the engine-{}", apexModel.getKey().getId(), engineWorkerKey);
 297         LOGGER.exit();
 298     }
 299
 300     /*
 301      * (non-Javadoc)
 302      *
 303      * @see org.onap.policy.apex.service.engine.runtime.EngineService#getState()
 304      */
 305     @Override
 306     public AxEngineState getState() {
 307         return engine.getState();
 308     }
 309
 310     /*
 311      * (non-Javadoc)
 312      *
 313      * @see org.onap.policy.apex.service.engine.runtime.EngineService#startAll()
 314      */
 315     @Override
 316     public void startAll() throws ApexException {
 317         start(this.getKey());
 318     }
 319
 320     /*
 321      * (non-Javadoc)
 322      *
 323      * @see
 324      * org.onap.policy.apex.service.engine.runtime.EngineService#start(org.onap.policy.apex.core.
 325      * model. concepts.AxArtifactKey)
 326      */
 327     @Override
 328     public void start(final AxArtifactKey engineKey) throws ApexException {
 329         LOGGER.entry(engineKey);
 330
 331         // Check if the key on the start request is correct
 332         if (!engineWorkerKey.equals(engineKey)) {
 333             LOGGER.warn(ENGINE_KEY_PREFIX + engineKey.getId() + BAD_KEY_MATCH_TAG + engineWorkerKey.getId()
 334                     + ENGINE_SUFFIX);
 335             throw new ApexException(ENGINE_KEY_PREFIX + engineKey.getId() + BAD_KEY_MATCH_TAG
 336                     + engineWorkerKey.getId() + ENGINE_SUFFIX);
 337         }
 338
 339         if (engine == null) {
 340             String message = ENGINE_FOR_KEY_PREFIX + engineWorkerKey.getId() + " is null";
 341             LOGGER.error(message);
 342             throw new ApexException(message);
 343         }
 344
 345         // Starts the event processing thread that handles incoming events
 346         if (processorThread != null && processorThread.isAlive()) {
 347             String message = ENGINE_FOR_KEY_PREFIX + engineWorkerKey.getId() + " is already running with state "
 348                     + getState();
 349             LOGGER.error(message);
 350             throw new ApexException(message);
 351         }
 352
 353         // Start the engine
 354         engine.start();
 355
 356         // Start a thread to process events for the engine
 357         processorThread = threadFactory.newThread(processor);
 358         processorThread.start();
 359
 360         LOGGER.exit(engineKey);
 361     }
 362
 363     /*
 364      * (non-Javadoc)
 365      *
 366      * @see org.onap.policy.apex.service.engine.runtime.EngineService#stop()
 367      */
 368     @Override
 369     public void stop() throws ApexException {
 370         stop(this.getKey());
 371     }
 372
 373     /*
 374      * (non-Javadoc)
 375      *
 376      * @see
 377      * org.onap.policy.apex.service.engine.runtime.EngineService#stop(org.onap.policy.apex.core.
 378      * model. concepts.AxArtifactKey)
 379      */
 380     @Override
 381     public void stop(final AxArtifactKey engineKey) throws ApexException {
 382         // Check if the key on the start request is correct
 383         if (!engineWorkerKey.equals(engineKey)) {
 384             LOGGER.warn(ENGINE_KEY_PREFIX + engineKey.getId() + BAD_KEY_MATCH_TAG + engineWorkerKey.getId()
 385                     + ENGINE_SUFFIX);
 386             throw new ApexException(ENGINE_KEY_PREFIX + engineKey.getId() + BAD_KEY_MATCH_TAG
 387                     + engineWorkerKey.getId() + ENGINE_SUFFIX);
 388         }
 389
 390         if (engine == null) {
 391             String message = ENGINE_FOR_KEY_PREFIX + engineWorkerKey.getId() + " is null";
 392             LOGGER.error(message);
 393             throw new ApexException(message);
 394         }
 395
 396         // Interrupt the worker to stop its thread
 397         if (processorThread == null || !processorThread.isAlive()) {
 398             processorThread = null;
 399
 400             LOGGER.warn(ENGINE_FOR_KEY_PREFIX + engineWorkerKey.getId() + " is already stopped with state "
 401                     + getState());
 402             return;
 403         }
 404
 405         // Interrupt the thread that is handling events toward the engine
 406         processorThread.interrupt();
 407         processorThread = null;
 408
 409         // Stop the engine
 410         engine.stop();
 411
 412         LOGGER.exit(engineKey);
 413     }
 414
 415     /*
 416      * (non-Javadoc)
 417      *
 418      * @see org.onap.policy.apex.service.engine.runtime.EngineService#clear()
 419      */
 420     @Override
 421     public void clear() throws ApexException {
 422         clear(this.getKey());
 423     }
 424
 425     /*
 426      * (non-Javadoc)
 427      *
 428      * @see
 429      * org.onap.policy.apex.service.engine.runtime.EngineService#clear(org.onap.policy.apex.core.
 430      * model. concepts.AxArtifactKey)
 431      */
 432     @Override
 433     public void clear(final AxArtifactKey engineKey) throws ApexException {
 434         // Check if the key on the start request is correct
 435         if (!engineWorkerKey.equals(engineKey)) {
 436             LOGGER.warn(ENGINE_KEY_PREFIX + engineKey.getId() + BAD_KEY_MATCH_TAG + engineWorkerKey.getId()
 437                     + ENGINE_SUFFIX);
 438             throw new ApexException(ENGINE_KEY_PREFIX + engineKey.getId() + BAD_KEY_MATCH_TAG
 439                     + engineWorkerKey.getId() + ENGINE_SUFFIX);
 440         }
 441
 442         if (engine == null) {
 443             LOGGER.error(ENGINE_FOR_KEY_PREFIX + engineWorkerKey.getId() + IS_NULL_SUFFIX);
 444             throw new ApexException(ENGINE_FOR_KEY_PREFIX + engineWorkerKey.getId() + IS_NULL_SUFFIX);
 445         }
 446
 447         // Interrupt the worker to stop its thread
 448         if (processorThread != null && !processorThread.isAlive()) {
 449             LOGGER.warn(ENGINE_FOR_KEY_PREFIX + engineWorkerKey.getId() + " is not stopped with state "
 450                     + getState());
 451             return;
 452         }
 453
 454         // Clear the engine
 455         engine.clear();
 456
 457         LOGGER.exit(engineKey);
 458     }
 459
 460     /*
 461      * (non-Javadoc)
 462      *
 463      * @see org.onap.policy.apex.service.engine.runtime.EngineService#isStarted()
 464      */
 465     @Override
 466     public boolean isStarted() {
 467         return isStarted(this.getKey());
 468     }
 469
 470     /*
 471      * (non-Javadoc)
 472      *
 473      * @see
 474      * org.onap.policy.apex.service.engine.runtime.EngineService#isStarted(org.onap.policy.apex.
 475      * model. basicmodel.concepts.AxArtifactKey)
 476      */
 477     @Override
 478     public boolean isStarted(final AxArtifactKey engineKey) {
 479         final AxEngineState engstate = getState();
 480         switch (engstate) {
 481             case STOPPED:
 482             case STOPPING:
 483             case UNDEFINED:
 484                 return false;
 485             case EXECUTING:
 486             case READY:
 487                 return processorThread != null && processorThread.isAlive() && !processorThread.isInterrupted();
 488             default:
 489                 break;
 490         }
 491         return false;
 492     }
 493
 494     /*
 495      * (non-Javadoc)
 496      *
 497      * @see org.onap.policy.apex.service.engine.runtime.EngineService#isStopped()
 498      */
 499     @Override
 500     public boolean isStopped() {
 501         return isStopped(this.getKey());
 502     }
 503
 504     /*
 505      * (non-Javadoc)
 506      *
 507      * @see
 508      * org.onap.policy.apex.service.engine.runtime.EngineService#isStopped(org.onap.policy.apex.
 509      * model. basicmodel.concepts.AxArtifactKey)
 510      */
 511     @Override
 512     public boolean isStopped(final AxArtifactKey engineKey) {
 513         final AxEngineState engstate = getState();
 514         switch (engstate) {
 515             case STOPPING:
 516             case UNDEFINED:
 517             case EXECUTING:
 518             case READY:
 519                 return false;
 520             case STOPPED:
 521                 return processorThread == null || !processorThread.isAlive();
 522             default:
 523                 break;
 524         }
 525         return false;
 526     }
 527
 528     /*
 529      * (non-Javadoc)
 530      *
 531      * @see org.onap.policy.apex.service.engine.runtime.EngineService#startPeriodicEvents(long)
 532      */
 533     @Override
 534     public void startPeriodicEvents(final long period) {
 535         throw new UnsupportedOperationException("startPeriodicEvents() call is not allowed on an Apex Engine Worker");
 536     }
 537
 538     /*
 539      * (non-Javadoc)
 540      *
 541      * @see org.onap.policy.apex.service.engine.runtime.EngineService#stopPeriodicEvents()
 542      */
 543     @Override
 544     public void stopPeriodicEvents() {
 545         throw new UnsupportedOperationException("stopPeriodicEvents() call is not allowed on an Apex Engine Worker");
 546     }
 547
 548     /*
 549      * (non-Javadoc)
 550      *
 551      * @see
 552      * org.onap.policy.apex.service.engine.runtime.EngineService#getStatus(org.onap.policy.apex.core
 553      * .model .concepts.AxArtifactKey)
 554      */
 555     @Override
 556     public String getStatus(final AxArtifactKey engineKey) {
 557         // Get the information from the engine that we want to return
 558         final AxEngineModel apexEngineModel = engine.getEngineStatus();
 559         apexEngineModel.getKeyInformation().generateKeyInfo(apexEngineModel);
 560
 561         // Convert that information into a string
 562         try {
 563             final ByteArrayOutputStream baOutputStream = new ByteArrayOutputStream();
 564             final ApexModelWriter<AxEngineModel> modelWriter = new ApexModelWriter<>(AxEngineModel.class);
 565             modelWriter.write(apexEngineModel, baOutputStream);
 566             return baOutputStream.toString();
 567         } catch (final Exception e) {
 568             LOGGER.warn("error outputting runtime information for engine {}", engineWorkerKey, e);
 569             return null;
 570         }
 571     }
 572
 573     /*
 574      * (non-Javadoc)
 575      *
 576      * @see
 577      * org.onap.policy.apex.service.engine.runtime.EngineService#getRuntimeInfo(org.onap.policy.apex
 578      * .core.model.concepts.AxArtifactKey)
 579      */
 580     @Override
 581     public String getRuntimeInfo(final AxArtifactKey engineKey) {
 582         // We'll build up the JSON string for runtime information bit by bit
 583         final StringBuilder runtimeJsonStringBuilder = new StringBuilder();
 584
 585         // Get the engine information
 586         final AxEngineModel engineModel = engine.getEngineStatus();
 587         final Map<AxArtifactKey, Map<String, Object>> engineContextAlbums = engine.getEngineContext();
 588
 589         // Use GSON to convert our context information into JSON
 590         final Gson gson = new GsonBuilder().setPrettyPrinting().create();
 591
 592         // Get context into a JSON string
 593         runtimeJsonStringBuilder.append("{\"TimeStamp\":");
 594         runtimeJsonStringBuilder.append(engineModel.getTimestamp());
 595         runtimeJsonStringBuilder.append(",\"State\":");
 596         runtimeJsonStringBuilder.append(engineModel.getState());
 597         runtimeJsonStringBuilder.append(",\"Stats\":");
 598         runtimeJsonStringBuilder.append(gson.toJson(engineModel.getStats()));
 599
 600         // Get context into a JSON string
 601         runtimeJsonStringBuilder.append(",\"ContextAlbums\":[");
 602
 603         boolean firstAlbum = true;
 604         for (final Entry<AxArtifactKey, Map<String, Object>> contextAlbumEntry : engineContextAlbums.entrySet()) {
 605             if (firstAlbum) {
 606                 firstAlbum = false;
 607             } else {
 608                 runtimeJsonStringBuilder.append(",");
 609             }
 610
 611             runtimeJsonStringBuilder.append("{\"AlbumKey\":");
 612             runtimeJsonStringBuilder.append(gson.toJson(contextAlbumEntry.getKey()));
 613             runtimeJsonStringBuilder.append(",\"AlbumContent\":[");
 614
 615
 616             // Get the schema helper to use to marshal context album objects to JSON
 617             final AxContextAlbum axContextAlbum =
 618                     ModelService.getModel(AxContextAlbums.class).get(contextAlbumEntry.getKey());
 619             SchemaHelper schemaHelper = null;
 620
 621             try {
 622                 // Get a schema helper to manage the translations between objects on the album map
 623                 // for this album
 624                 schemaHelper = new SchemaHelperFactory().createSchemaHelper(axContextAlbum.getKey(),
 625                         axContextAlbum.getItemSchema());
 626             } catch (final ContextRuntimeException e) {
 627                 final String resultString =
 628                         "could not find schema helper to marshal context album \"" + axContextAlbum + "\" to JSON";
 629                 LOGGER.warn(resultString, e);
 630
 631                 // End of context album entry
 632                 runtimeJsonStringBuilder.append(resultString);
 633                 runtimeJsonStringBuilder.append("]}");
 634
 635                 continue;
 636             }
 637
 638             boolean firstEntry = true;
 639             for (final Entry<String, Object> contextEntry : contextAlbumEntry.getValue().entrySet()) {
 640                 if (firstEntry) {
 641                     firstEntry = false;
 642                 } else {
 643                     runtimeJsonStringBuilder.append(",");
 644                 }
 645                 runtimeJsonStringBuilder.append("{\"EntryName\":");
 646                 runtimeJsonStringBuilder.append(gson.toJson(contextEntry.getKey()));
 647                 runtimeJsonStringBuilder.append(",\"EntryContent\":");
 648                 runtimeJsonStringBuilder.append(gson.toJson(schemaHelper.marshal2String(contextEntry.getValue())));
 649
 650                 // End of context entry
 651                 runtimeJsonStringBuilder.append("}");
 652             }
 653
 654             // End of context album entry
 655             runtimeJsonStringBuilder.append("]}");
 656         }
 657
 658         runtimeJsonStringBuilder.append("]}");
 659
 660         // Tidy up the JSON string
 661         final JsonParser jsonParser = new JsonParser();
 662         final JsonElement jsonElement = jsonParser.parse(runtimeJsonStringBuilder.toString());
 663         final String tidiedRuntimeString = gson.toJson(jsonElement);
 664
 665         LOGGER.debug("runtime information={}", tidiedRuntimeString);
 666
 667         return tidiedRuntimeString;
 668     }
 669
 670     /**
 671      * This is an event processor thread, this class decouples the events handling logic from core
 672      * business logic. This class runs its own thread and continuously querying the blocking queue
 673      * for the events that have been sent to the worker for processing by the Apex engine.
 674      *
 675      * @author Liam Fallon (liam.fallon@ericsson.com)
 676      */
 677     private class EventProcessor implements Runnable {
 678         private final boolean debugEnabled = LOGGER.isDebugEnabled();
 679         // the events queue
 680         private BlockingQueue<ApexEvent> eventProcessingQueue = null;
 681
 682         /**
 683          * Constructor accepts {@link ApexEngine} and {@link BlockingQueue} type objects.
 684          *
 685          * @param eventProcessingQueue is reference of {@link BlockingQueue} which contains trigger
 686          *        events.
 687          */
 688         EventProcessor(final BlockingQueue<ApexEvent> eventProcessingQueue) {
 689             this.eventProcessingQueue = eventProcessingQueue;
 690         }
 691
 692         /*
 693          * (non-Javadoc)
 694          *
 695          * @see java.lang.Runnable#run()
 696          */
 697         @Override
 698         public void run() {
 699             LOGGER.debug("Engine {} processing ... ", engineWorkerKey);
 700
 701             // Take events from the event processing queue of the worker and pass them to the engine
 702             // for processing
 703             boolean stopFlag = false;
 704             while (!processorThread.isInterrupted() && ! stopFlag) {
 705                 ApexEvent event = null;
 706                 try {
 707                     event = eventProcessingQueue.take();
 708                 } catch (final InterruptedException e) {
 709                     // restore the interrupt status
 710                     Thread.currentThread().interrupt();
 711                     LOGGER.debug("Engine {} processing interrupted ", engineWorkerKey);
 712                     break;
 713                 }
 714
 715                 try {
 716                     if (event != null) {
 717                         debugEventIfDebugEnabled(event);
 718
 719                         final EnEvent enevent = apexEnEventConverter.fromApexEvent(event);
 720                         engine.handleEvent(enevent);
 721                     }
 722                 } catch (final ApexException e) {
 723                     LOGGER.warn("Engine {} failed to process event {}", engineWorkerKey, event.toString(), e);
 724                 } catch (final Exception e) {
 725                     LOGGER.warn("Engine {} terminated processing event {}", engineWorkerKey, event.toString(), e);
 726                     stopFlag = true;
 727                 }
 728             }
 729             LOGGER.debug("Engine {} completed processing", engineWorkerKey);
 730         }
 731
 732         /**
 733          * Debug the event if debug is enabled.
 734          *
 735          * @param event the event to debug
 736          */
 737         private void debugEventIfDebugEnabled(ApexEvent event) {
 738             if (debugEnabled) {
 739                 LOGGER.debug("Trigger Event {} forwarded to the Apex engine", event);
 740             }
 741         }
 742     }
 743 }