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.engdep;
  22
  23 import com.google.common.eventbus.Subscribe;
  24
  25 import java.util.Collection;
  26 import java.util.List;
  27 import java.util.concurrent.BlockingQueue;
  28 import java.util.concurrent.LinkedBlockingDeque;
  29 import java.util.concurrent.TimeUnit;
  30
  31 import org.java_websocket.WebSocket;
  32 import org.onap.policy.apex.core.infrastructure.messaging.MessageHolder;
  33 import org.onap.policy.apex.core.infrastructure.messaging.MessageListener;
  34 import org.onap.policy.apex.core.infrastructure.messaging.impl.ws.messageblock.MessageBlock;
  35 import org.onap.policy.apex.core.infrastructure.messaging.util.MessagingUtils;
  36 import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities;
  37 import org.onap.policy.apex.core.protocols.Message;
  38 import org.onap.policy.apex.core.protocols.engdep.EngDepAction;
  39 import org.onap.policy.apex.core.protocols.engdep.messages.EngineServiceInfoResponse;
  40 import org.onap.policy.apex.core.protocols.engdep.messages.GetEngineInfo;
  41 import org.onap.policy.apex.core.protocols.engdep.messages.GetEngineServiceInfo;
  42 import org.onap.policy.apex.core.protocols.engdep.messages.GetEngineStatus;
  43 import org.onap.policy.apex.core.protocols.engdep.messages.Response;
  44 import org.onap.policy.apex.core.protocols.engdep.messages.StartEngine;
  45 import org.onap.policy.apex.core.protocols.engdep.messages.StartPeriodicEvents;
  46 import org.onap.policy.apex.core.protocols.engdep.messages.StopEngine;
  47 import org.onap.policy.apex.core.protocols.engdep.messages.StopPeriodicEvents;
  48 import org.onap.policy.apex.core.protocols.engdep.messages.UpdateModel;
  49 import org.onap.policy.apex.model.basicmodel.concepts.ApexException;
  50 import org.onap.policy.apex.model.basicmodel.concepts.AxArtifactKey;
  51 import org.onap.policy.apex.service.engine.runtime.EngineService;
  52 import org.slf4j.ext.XLogger;
  53 import org.slf4j.ext.XLoggerFactory;
  54
  55 /**
  56  * The listener interface for receiving engDepMessage events. The class that is interested in processing a engDepMessage
  57  * event implements this interface, and the object created with that class is registered with a component using the
  58  * component's <code>addEngDepMessageListener</code> method. When the engDepMessage event occurs, that object's
  59  * appropriate method is invoked.
  60  *
  61  * <p>This class uses a queue to buffer incoming messages. When the listener is called, it places the incoming message
  62  * on the queue. A thread runs which removes the messages from the queue and forwards them to the Apex engine.
  63  *
  64  * @author Sajeevan Achuthan (sajeevan.achuthan@ericsson.com)
  65  */
  66 public class EngDepMessageListener implements MessageListener<Message>, Runnable {
  67     private static final int LISTENER_STOP_WAIT_INTERVAL = 10;
  68
  69     private static final XLogger LOGGER = XLoggerFactory.getXLogger(EngDepMessageListener.class);
  70
  71     // The timeout to wait between queue poll timeouts in milliseconds
  72     private static final long QUEUE_POLL_TIMEOUT = 50;
  73
  74     // The Apex service itself
  75     private final EngineService apexService;
  76
  77     // The message listener thread and stopping flag
  78     private Thread messageListenerThread;
  79     private boolean stopOrderedFlag = false;
  80
  81     // The message queue is used to hold messages prior to forwarding to Apex
  82     private final BlockingQueue<MessageBlock<Message>> messageQueue = new LinkedBlockingDeque<>();
  83
  84     /**
  85      * Instantiates a new EngDep message listener for listening for messages coming in from the Deployment client. The
  86      * <code>apexService</code> is the Apex service to send the messages onto.
  87      *
  88      * @param apexService the Apex engine service
  89      */
  90     protected EngDepMessageListener(final EngineService apexService) {
  91         this.apexService = apexService;
  92     }
  93
  94     /**
  95      * This method is an implementation of the message listener. It receives a message and places it on the queue for
  96      * processing by the message listening thread.
  97      *
  98      * @param data the data
  99      * @see org.onap.policy.apex.core.infrastructure.messaging.MessageListener#onMessage
 100      *      (org.onap.policy.apex.core.infrastructure.messaging.impl.ws.data.Data)
 101      */
 102     @Subscribe
 103     @Override
 104     public void onMessage(final MessageBlock<Message> data) {
 105         if (LOGGER.isDebugEnabled()) {
 106             LOGGER.debug("message received from client application {} port {}",
 107                             data.getConnection().getRemoteSocketAddress().getAddress(),
 108                             data.getConnection().getRemoteSocketAddress().getPort());
 109         }
 110         messageQueue.add(data);
 111     }
 112
 113     /*
 114      * (non-Javadoc)
 115      *
 116      * @see org.onap.policy.apex.core.infrastructure.messaging.MessageListener#onMessage(java.lang. String)
 117      */
 118     @Override
 119     public void onMessage(final String messageString) {
 120         throw new UnsupportedOperationException("String messages are not supported on the EngDep protocol");
 121     }
 122
 123     /**
 124      * This method gets a new message listening thread from the thread factory and starts it.
 125      */
 126     public void startProcessorThread() {
 127         LOGGER.entry();
 128         messageListenerThread = new Thread(this);
 129         messageListenerThread.setDaemon(true);
 130         messageListenerThread.start();
 131         LOGGER.exit();
 132     }
 133
 134     /**
 135      * Stops the message listening threads.
 136      */
 137     public void stopProcessorThreads() {
 138         LOGGER.entry();
 139         stopOrderedFlag = true;
 140
 141         while (messageListenerThread.isAlive()) {
 142             ThreadUtilities.sleep(LISTENER_STOP_WAIT_INTERVAL);
 143         }
 144         LOGGER.exit();
 145     }
 146
 147     /**
 148      * Runs the message listening thread. Here, the messages come in on the message queue and are processed one by one
 149      */
 150     @Override
 151     public void run() {
 152         // Take messages off the queue and forward them to the Apex engine
 153         while (messageListenerThread.isAlive() && !stopOrderedFlag) {
 154             pollAndHandleMessage();
 155         }
 156     }
 157
 158     /**
 159      * Poll the queue for a message and handle that message.
 160      */
 161     private void pollAndHandleMessage() {
 162         try {
 163             final MessageBlock<Message> data = messageQueue.poll(QUEUE_POLL_TIMEOUT, TimeUnit.MILLISECONDS);
 164             if (data != null) {
 165                 final List<Message> messages = data.getMessages();
 166                 for (final Message message : messages) {
 167                     handleMessage(message, data.getConnection());
 168                 }
 169             }
 170         } catch (final InterruptedException e) {
 171             // restore the interrupt status
 172             Thread.currentThread().interrupt();
 173             LOGGER.debug("message listener execution has been interrupted");
 174         }
 175     }
 176
 177     /**
 178      * This method handles EngDep messages as they come in. It uses the inevitable switch statement to handle the
 179      * messages.
 180      *
 181      * @param message the incoming EngDep message
 182      * @param webSocket the web socket on which the message came in
 183      */
 184     private void handleMessage(final Message message, final WebSocket webSocket) {
 185         LOGGER.entry(webSocket.getRemoteSocketAddress().toString());
 186         if (message.getAction() == null) {
 187             // This is a response message
 188             return;
 189         }
 190
 191         try {
 192             LOGGER.debug("Manager action {} being applied to engine", message.getAction());
 193
 194             // Get and check the incoming action for validity
 195             EngDepAction enDepAction = null;
 196             if (message.getAction() instanceof EngDepAction) {
 197                 enDepAction = (EngDepAction) message.getAction();
 198             } else {
 199                 throw new ApexException(message.getAction().getClass().getName()
 200                                 + "action on received message invalid, action must be of type \"EnDepAction\"");
 201             }
 202
 203             handleIncomingMessages(message, webSocket, enDepAction);
 204         } catch (final ApexException e) {
 205             LOGGER.warn("apex failed to execute message", e);
 206             sendReply(webSocket, message, false, e.getCascadedMessage());
 207         } catch (final Exception e) {
 208             LOGGER.warn("system failure executing message", e);
 209             sendReply(webSocket, message, false, e.getMessage());
 210         }
 211         LOGGER.exit();
 212     }
 213
 214     /**
 215      * Handle incoming EngDep messages.
 216      *
 217      * @param message the incoming message
 218      * @param webSocket the web socket the message came in on
 219      * @param enDepAction the action from the message
 220      * @throws ApexException on message handling errors
 221      */
 222     private void handleIncomingMessages(final Message message, final WebSocket webSocket, EngDepAction enDepAction)
 223                     throws ApexException {
 224         // Handle each incoming message using the inevitable switch statement for the EngDep
 225         // protocol
 226         switch (enDepAction) {
 227             case GET_ENGINE_SERVICE_INFO:
 228                 handleGetEngineServiceInfoMessage(message, webSocket);
 229                 break;
 230
 231             case UPDATE_MODEL:
 232                 handleUpdateModelMessage(message, webSocket);
 233                 break;
 234
 235             case START_ENGINE:
 236                 handleStartEngineMessage(message, webSocket);
 237                 break;
 238
 239             case STOP_ENGINE:
 240                 handleStopEngineMessage(message, webSocket);
 241                 break;
 242
 243             case START_PERIODIC_EVENTS:
 244                 handleStartPeriodicEventsMessage(message, webSocket);
 245                 break;
 246
 247             case STOP_PERIODIC_EVENTS:
 248                 handleStopPeriodicEventsMessage(message, webSocket);
 249                 break;
 250
 251             case GET_ENGINE_STATUS:
 252                 handleEngineStatusMessage(message, webSocket);
 253                 break;
 254
 255             case GET_ENGINE_INFO:
 256                 handleEngineInfoMessage(message, webSocket);
 257                 break;
 258
 259             case RESPONSE:
 260                 throw new ApexException("RESPONSE action on received message not handled by engine");
 261
 262             default:
 263                 break;
 264         }
 265     }
 266
 267     /**
 268      * Handle the get engine service information message.
 269      *
 270      * @param message the message
 271      * @param webSocket the web socket that the message came on
 272      * @throws ApexException on message handling exceptions
 273      */
 274     private void handleGetEngineServiceInfoMessage(final Message message, final WebSocket webSocket) {
 275         final GetEngineServiceInfo engineServiceInformationMessage = (GetEngineServiceInfo) message;
 276         LOGGER.debug("getting engine service information for engine service " + apexService.getKey().getId()
 277                         + " . . .");
 278         // Send a reply with the engine service information
 279         sendServiceInfoReply(webSocket, engineServiceInformationMessage, apexService.getKey(),
 280                         apexService.getEngineKeys(), apexService.getApexModelKey());
 281         LOGGER.debug("returned engine service information for engine service "
 282                         + apexService.getKey().getId());
 283     }
 284
 285     /**
 286      * Handle the update model message.
 287      *
 288      * @param message the message
 289      * @param webSocket the web socket that the message came on
 290      * @throws ApexException on message handling exceptions
 291      */
 292     private void handleUpdateModelMessage(final Message message, final WebSocket webSocket) throws ApexException {
 293         final UpdateModel updateModelMessage = (UpdateModel) message;
 294         LOGGER.debug("updating model in engine {} . . .", updateModelMessage.getTarget().getId());
 295         // Update the model
 296         apexService.updateModel(updateModelMessage.getTarget(), updateModelMessage.getMessageData(),
 297                         updateModelMessage.isForceInstall());
 298         // Send a reply indicating the message action worked
 299         sendReply(webSocket, updateModelMessage, true,
 300                         "updated model in engine " + updateModelMessage.getTarget().getId());
 301         LOGGER.debug("updated model in engine service {}", updateModelMessage.getTarget().getId());
 302     }
 303
 304     /**
 305      * Handle the start engine message.
 306      *
 307      * @param message the message
 308      * @param webSocket the web socket that the message came on
 309      * @throws ApexException on message handling exceptions
 310      */
 311     private void handleStartEngineMessage(final Message message, final WebSocket webSocket) throws ApexException {
 312         final StartEngine startEngineMessage = (StartEngine) message;
 313         LOGGER.debug("starting engine {} . . .", startEngineMessage.getTarget().getId());
 314         // Start the engine
 315         apexService.start(startEngineMessage.getTarget());
 316         // Send a reply indicating the message action worked
 317         sendReply(webSocket, startEngineMessage, true,
 318                         "started engine " + startEngineMessage.getTarget().getId());
 319         LOGGER.debug("started engine {}", startEngineMessage.getTarget().getId());
 320     }
 321
 322     /**
 323      * Handle the stop engine message.
 324      *
 325      * @param message the message
 326      * @param webSocket the web socket that the message came on
 327      * @throws ApexException on message handling exceptions
 328      */
 329     private void handleStopEngineMessage(final Message message, final WebSocket webSocket) throws ApexException {
 330         final StopEngine stopEngineMessage = (StopEngine) message;
 331         LOGGER.debug("stopping engine {} . . .", stopEngineMessage.getTarget().getId());
 332         // Stop the engine
 333         apexService.stop(stopEngineMessage.getTarget());
 334         // Send a reply indicating the message action worked
 335         sendReply(webSocket, stopEngineMessage, true,
 336                         "stopped engine " + stopEngineMessage.getTarget().getId());
 337         LOGGER.debug("stopping engine {}", stopEngineMessage.getTarget().getId());
 338     }
 339
 340     /**
 341      * Handle the start periodic events message.
 342      *
 343      * @param message the message
 344      * @param webSocket the web socket that the message came on
 345      * @throws ApexException on message handling exceptions
 346      */
 347     private void handleStartPeriodicEventsMessage(final Message message, final WebSocket webSocket)
 348                     throws ApexException {
 349         final StartPeriodicEvents startPeriodicEventsMessage = (StartPeriodicEvents) message;
 350         LOGGER.debug("starting periodic events on engine {} . . .",
 351                         startPeriodicEventsMessage.getTarget().getId());
 352         // Start periodic events with the period specified in the message
 353         final Long period = Long.parseLong(startPeriodicEventsMessage.getMessageData());
 354         apexService.startPeriodicEvents(period);
 355         // Send a reply indicating the message action worked
 356         String periodicStartedMessage = "started periodic events on engine "
 357                         + startPeriodicEventsMessage.getTarget().getId() + " with period " + period;
 358         sendReply(webSocket, startPeriodicEventsMessage, true, periodicStartedMessage);
 359         LOGGER.debug(periodicStartedMessage);
 360     }
 361
 362     /**
 363      * Handle the stop periodic events message.
 364      *
 365      * @param message the message
 366      * @param webSocket the web socket that the message came on
 367      * @throws ApexException on message handling exceptions
 368      */
 369     private void handleStopPeriodicEventsMessage(final Message message, final WebSocket webSocket)
 370                     throws ApexException {
 371         final StopPeriodicEvents stopPeriodicEventsMessage = (StopPeriodicEvents) message;
 372         LOGGER.debug("stopping periodic events on engine {} . . .",
 373                         stopPeriodicEventsMessage.getTarget().getId());
 374         // Stop periodic events
 375         apexService.stopPeriodicEvents();
 376         // Send a reply indicating the message action worked
 377         sendReply(webSocket, stopPeriodicEventsMessage, true, "stopped periodic events on engine "
 378                         + stopPeriodicEventsMessage.getTarget().getId());
 379         LOGGER.debug("stopped periodic events on engine " + stopPeriodicEventsMessage.getTarget().getId());
 380     }
 381
 382     /**
 383      * Handle the engine status message.
 384      *
 385      * @param message the message
 386      * @param webSocket the web socket that the message came on
 387      * @throws ApexException on message handling exceptions
 388      */
 389     private void handleEngineStatusMessage(final Message message, final WebSocket webSocket) throws ApexException {
 390         final GetEngineStatus getEngineStatusMessage = (GetEngineStatus) message;
 391         LOGGER.debug("getting status for engine{} . . .", getEngineStatusMessage.getTarget().getId());
 392         // Send a reply with the engine status
 393         sendReply(webSocket, getEngineStatusMessage, true,
 394                         apexService.getStatus(getEngineStatusMessage.getTarget()));
 395         LOGGER.debug("returned status for engine {}", getEngineStatusMessage.getTarget().getId());
 396     }
 397
 398     /**
 399      * Handle the engine information message.
 400      *
 401      * @param message the message
 402      * @param webSocket the web socket that the message came on
 403      * @throws ApexException on message handling exceptions
 404      */
 405     private void handleEngineInfoMessage(final Message message, final WebSocket webSocket) throws ApexException {
 406         final GetEngineInfo getEngineInfo = (GetEngineInfo) message;
 407         LOGGER.debug("getting runtime information for engine {} . . .", getEngineInfo.getTarget().getId());
 408         // Send a reply with the engine runtime information
 409         sendReply(webSocket, getEngineInfo, true, apexService.getRuntimeInfo(getEngineInfo.getTarget()));
 410         LOGGER.debug("returned runtime information for engine {}", getEngineInfo.getTarget().getId());
 411     }
 412
 413     /**
 414      * Send the Response message to the client.
 415      *
 416      * @param client the client to which to send the response message
 417      * @param requestMessage the message to which we are responding
 418      * @param result the result indicating success or failure
 419      * @param messageData the message data
 420      */
 421     private void sendReply(final WebSocket client, final Message requestMessage, final boolean result,
 422                     final String messageData) {
 423         LOGGER.entry(result, messageData);
 424
 425         if (client == null || !client.isOpen()) {
 426             LOGGER.debug("error sending reply {}, client has disconnected", requestMessage.getAction());
 427             return;
 428         }
 429
 430         String replyString = "sending " + requestMessage.getAction() + " to web socket "
 431                         + client.getRemoteSocketAddress().toString();
 432         LOGGER.debug(replyString);
 433
 434         final Response responseMessage = new Response(requestMessage.getTarget(), result, requestMessage);
 435         responseMessage.setMessageData(messageData);
 436
 437         final MessageHolder<Message> messageHolder = new MessageHolder<>(MessagingUtils.getHost());
 438         messageHolder.addMessage(responseMessage);
 439         client.send(MessagingUtils.serializeObject(messageHolder));
 440
 441         LOGGER.exit();
 442     }
 443
 444     /**
 445      * Send the EngineServiceInfoResponse message to the client.
 446      *
 447      * @param client the client to which to send the response message
 448      * @param requestMessage the message to which we are responding
 449      * @param engineServiceKey The key of this engine service
 450      * @param engineKeyCollection The keys of the engines in this engine service
 451      * @param apexModelKey the apex model key
 452      */
 453     private void sendServiceInfoReply(final WebSocket client, final Message requestMessage,
 454                     final AxArtifactKey engineServiceKey, final Collection<AxArtifactKey> engineKeyCollection,
 455                     final AxArtifactKey apexModelKey) {
 456         LOGGER.entry();
 457         String sendingMessage = "sending " + requestMessage.getAction() + " to web socket "
 458                         + client.getRemoteSocketAddress().toString();
 459         LOGGER.debug(sendingMessage);
 460
 461         final EngineServiceInfoResponse responseMessage = new EngineServiceInfoResponse(requestMessage.getTarget(),
 462                         true, requestMessage);
 463         responseMessage.setMessageData("engine service information");
 464         responseMessage.setEngineServiceKey(engineServiceKey);
 465         responseMessage.setEngineKeyArray(engineKeyCollection);
 466         responseMessage.setApexModelKey(apexModelKey);
 467
 468         final MessageHolder<Message> messageHolder = new MessageHolder<>(MessagingUtils.getHost());
 469         messageHolder.addMessage(responseMessage);
 470         client.send(MessagingUtils.serializeObject(messageHolder));
 471
 472         LOGGER.exit();
 473     }
 474 }