aboutsummaryrefslogtreecommitdiffstats
path: root/policy-management/src/main/java/org/openecomp/policy/drools/system/PolicyEngine.java
diff options
context:
space:
mode:
Diffstat (limited to 'policy-management/src/main/java/org/openecomp/policy/drools/system/PolicyEngine.java')
-rw-r--r--policy-management/src/main/java/org/openecomp/policy/drools/system/PolicyEngine.java1182
1 files changed, 1182 insertions, 0 deletions
diff --git a/policy-management/src/main/java/org/openecomp/policy/drools/system/PolicyEngine.java b/policy-management/src/main/java/org/openecomp/policy/drools/system/PolicyEngine.java
new file mode 100644
index 00000000..33f2a098
--- /dev/null
+++ b/policy-management/src/main/java/org/openecomp/policy/drools/system/PolicyEngine.java
@@ -0,0 +1,1182 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * policy-management
+ * ================================================================================
+ * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.openecomp.policy.drools.system;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import org.openecomp.policy.common.logging.eelf.MessageCodes;
+import org.openecomp.policy.common.logging.flexlogger.FlexLogger;
+import org.openecomp.policy.common.logging.flexlogger.Logger;
+import org.openecomp.policy.drools.controller.DroolsController;
+import org.openecomp.policy.drools.core.FeatureAPI;
+import org.openecomp.policy.drools.core.jmx.PdpJmxListener;
+import org.openecomp.policy.drools.event.comm.Topic;
+import org.openecomp.policy.drools.event.comm.Topic.CommInfrastructure;
+import org.openecomp.policy.drools.event.comm.TopicEndpoint;
+import org.openecomp.policy.drools.event.comm.TopicListener;
+import org.openecomp.policy.drools.event.comm.TopicSink;
+import org.openecomp.policy.drools.event.comm.TopicSource;
+import org.openecomp.policy.drools.http.server.HttpServletServer;
+import org.openecomp.policy.drools.persistence.SystemPersistence;
+import org.openecomp.policy.drools.properties.Lockable;
+import org.openecomp.policy.drools.properties.PolicyProperties;
+import org.openecomp.policy.drools.properties.Startable;
+import org.openecomp.policy.drools.protocol.coders.EventProtocolCoder;
+import org.openecomp.policy.drools.protocol.configuration.ControllerConfiguration;
+import org.openecomp.policy.drools.protocol.configuration.PdpdConfiguration;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+
+/**
+ * Policy Engine, the top abstraction for the Drools PDP Policy Engine.
+ * It abstracts away a Drools PDP Engine from management purposes.
+ * This is the best place to looking at the code from a top down approach.
+ * Other managed entities can be obtained from the PolicyEngine, hierarchically.
+ * <br>
+ * PolicyEngine 1 --- * PolicyController 1 --- 1 DroolsController 1 --- 1 PolicyContainer 1 --- * PolicySession
+ * <br>
+ * PolicyEngine 1 --- 1 TopicEndpointManager 1 -- * TopicReader 1 --- 1 UebTopicReader
+ * <br>
+ * PolicyEngine 1 --- 1 TopicEndpointManager 1 -- * TopicReader 1 --- 1 DmaapTopicReader
+ * <br>
+ * PolicyEngine 1 --- 1 TopicEndpointManager 1 -- * TopicWriter 1 --- 1 DmaapTopicWriter
+ * <br>
+ * PolicyEngine 1 --- 1 TopicEndpointManager 1 -- * TopicReader 1 --- 1 RestTopicReader
+ * <br>
+ * PolicyEngine 1 --- 1 TopicEndpointManager 1 -- * TopicWriter 1 --- 1 RestTopicWriter
+ * <br>
+ * PolicyEngine 1 --- 1 ManagementServer
+ */
+public interface PolicyEngine extends Startable, Lockable, TopicListener {
+
+ /**
+ * Default Config Server Port
+ */
+ public static final int CONFIG_SERVER_DEFAULT_PORT = 9696;
+
+ /**
+ * Default Config Server Hostname
+ */
+ public static final String CONFIG_SERVER_DEFAULT_HOST = "localhost";
+
+ /**
+ * configure the policy engine according to the given properties
+ *
+ * @param properties Policy Engine properties
+ * @throws IllegalArgumentException when invalid or insufficient
+ * properties are provided
+ */
+ public void configure(Properties properties) throws IllegalArgumentException;
+
+ /**
+ * registers a new Policy Controller with the Policy Engine
+ * initialized per properties.
+ *
+ * @param controller name
+ * @param properties properties to initialize the Policy Controller
+ * @throws IllegalArgumentException when invalid or insufficient
+ * properties are provided
+ * @throws IllegalStateException when the engine is in a state where
+ * this operation is not permitted.
+ * @return the newly instantiated Policy Controller
+ */
+ public PolicyController createPolicyController(String name, Properties properties)
+ throws IllegalArgumentException, IllegalStateException;
+
+ /**
+ * updates the Policy Engine with the given configuration
+ *
+ * @param configuration the configuration
+ * @return success or failure
+ * @throws IllegalArgumentException if invalid argument provided
+ * @throws IllegalStateException if the system is in an invalid state
+ */
+ public boolean configure(PdpdConfiguration configuration)
+ throws IllegalArgumentException, IllegalStateException;
+
+ /**
+ * updates a set of Policy Controllers with configuration information
+ *
+ * @param configuration
+ * @return
+ * @throws IllegalArgumentException
+ * @throws IllegalStateException
+ */
+ public List<PolicyController> updatePolicyControllers(List<ControllerConfiguration> configuration)
+ throws IllegalArgumentException, IllegalStateException;
+
+ /**
+ * updates an already existing Policy Controller with configuration information
+ *
+ * @param configuration configuration
+ *
+ * @return the updated Policy Controller
+ * @throws IllegalArgumentException in the configuration is invalid
+ * @throws IllegalStateException if the controller is in a bad state
+ * @throws Exception any other reason
+ */
+ public PolicyController updatePolicyController(ControllerConfiguration configuration)
+ throws Exception;
+
+ /**
+ * removes the Policy Controller identified by its name from the Policy Engine
+ *
+ * @param name name of the Policy Controller
+ * @return the removed Policy Controller
+ */
+ public void removePolicyController(String name);
+
+ /**
+ * removes a Policy Controller from the Policy Engine
+ * @param controller the Policy Controller to remove from the Policy Engine
+ */
+ public void removePolicyController(PolicyController controller);
+
+ /**
+ * returns a list of the available Policy Controllers
+ *
+ * @return list of Policy Controllers
+ */
+ public List<PolicyController> getPolicyControllers();
+
+ /**
+ * get unmanaged sources
+ *
+ * @return unmanaged sources
+ */
+ public List<TopicSource> getSources();
+
+ /**
+ * get unmanaged sinks
+ *
+ * @return unmanaged sinks
+ */
+ public List<TopicSink> getSinks();
+
+ /**
+ * get unmmanaged http servers list
+ * @return http servers
+ */
+ public List<HttpServletServer> getHttpServers();
+
+ /**
+ * get properties configuration
+ *
+ * @return properties objects
+ */
+ public Properties getProperties();
+
+ /**
+ * Attempts the dispatching of an "event" object
+ *
+ * @param topic topic
+ * @param event the event object to send
+ *
+ * @return true if successful, false if a failure has occurred.
+ * @throws IllegalArgumentException when invalid or insufficient
+ * properties are provided
+ * @throws IllegalStateException when the engine is in a state where
+ * this operation is not permitted (ie. locked or stopped).
+ */
+ public boolean deliver(String topic, Object event)
+ throws IllegalArgumentException, IllegalStateException;
+
+ /**
+ * Attempts the dispatching of an "event" object over communication
+ * infrastructure "busType"
+ *
+ * @param eventBus Communication infrastructure identifier
+ * @param topic topic
+ * @param event the event object to send
+ *
+ * @return true if successful, false if a failure has occurred.
+ * @throws IllegalArgumentException when invalid or insufficient
+ * properties are provided
+ * @throws IllegalStateException when the engine is in a state where
+ * this operation is not permitted (ie. locked or stopped).
+ * @throws UnsupportedOperationException when the engine cannot deliver due
+ * to the functionality missing (ie. communication infrastructure
+ * not supported.
+ */
+ public boolean deliver(String busType, String topic, Object event)
+ throws IllegalArgumentException, IllegalStateException,
+ UnsupportedOperationException;
+
+ /**
+ * Attempts the dispatching of an "event" object over communication
+ * infrastructure "busType"
+ *
+ * @param eventBus Communication infrastructure enum
+ * @param topic topic
+ * @param event the event object to send
+ *
+ * @return true if successful, false if a failure has occurred.
+ * @throws IllegalArgumentException when invalid or insufficient
+ * properties are provided
+ * @throws IllegalStateException when the engine is in a state where
+ * this operation is not permitted (ie. locked or stopped).
+ * @throws UnsupportedOperationException when the engine cannot deliver due
+ * to the functionality missing (ie. communication infrastructure
+ * not supported.
+ */
+ public boolean deliver(CommInfrastructure busType, String topic, Object event)
+ throws IllegalArgumentException, IllegalStateException,
+ UnsupportedOperationException;
+
+ /**
+ * Attempts delivering of an String over communication
+ * infrastructure "busType"
+ *
+ * @param eventBus Communication infrastructure identifier
+ * @param topic topic
+ * @param event the event object to send
+ *
+ * @return true if successful, false if a failure has occurred.
+ * @throws IllegalArgumentException when invalid or insufficient
+ * properties are provided
+ * @throws IllegalStateException when the engine is in a state where
+ * this operation is not permitted (ie. locked or stopped).
+ * @throws UnsupportedOperationException when the engine cannot deliver due
+ * to the functionality missing (ie. communication infrastructure
+ * not supported.
+ */
+ public boolean deliver(CommInfrastructure busType, String topic,
+ String event)
+ throws IllegalArgumentException, IllegalStateException,
+ UnsupportedOperationException;
+
+ /**
+ * Invoked when the host goes into the active state.
+ */
+ public void activate();
+
+ /**
+ * Invoked when the host goes into the standby state.
+ */
+ public void deactivate();
+
+ /**
+ * get policy controller names
+ *
+ * @return list of controller names
+ */
+ public List<String> getControllers();
+
+ /**
+ * Policy Engine Manager
+ */
+ public final static PolicyEngine manager = new PolicyEngineManager();
+}
+
+/**
+ * Policy Engine Manager Implementation
+ */
+class PolicyEngineManager implements PolicyEngine {
+ /**
+ * logger
+ */
+ private static Logger logger = FlexLogger.getLogger(PolicyEngineManager.class);
+
+ /**
+ * Is the Policy Engine running?
+ */
+ protected boolean alive = false;
+
+ /**
+ * Is the engine locked?
+ */
+ protected boolean locked = false;
+
+ /**
+ * Properties used to initialize the engine
+ */
+ protected Properties properties;
+
+ /**
+ * Policy Engine Sources
+ */
+ protected List<? extends TopicSource> sources = new ArrayList<>();
+
+ /**
+ * Policy Engine Sinks
+ */
+ protected List<? extends TopicSink> sinks = new ArrayList<>();
+
+ /**
+ * Policy Engine HTTP Servers
+ */
+ protected List<HttpServletServer> httpServers = new ArrayList<HttpServletServer>();
+
+ protected Gson decoder = new GsonBuilder().disableHtmlEscaping().create();
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void configure(Properties properties) throws IllegalArgumentException {
+
+ if (properties == null) {
+ logger.warn("No properties provided");
+ throw new IllegalArgumentException("No properties provided");
+ }
+
+ this.properties = properties;
+
+ try {
+ this.sources = TopicEndpoint.manager.addTopicSources(properties);
+ for (TopicSource source: this.sources) {
+ source.register(this);
+ }
+ } catch (Exception e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, "PolicyEngine", "configure");
+ }
+
+ try {
+ this.sinks = TopicEndpoint.manager.addTopicSinks(properties);
+ } catch (IllegalArgumentException e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, "PolicyEngine", "configure");
+ }
+
+ try {
+ this.httpServers = HttpServletServer.factory.build(properties);
+ } catch (IllegalArgumentException e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, "PolicyEngine", "configure");
+ }
+
+ return;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public PolicyController createPolicyController(String name, Properties properties)
+ throws IllegalArgumentException, IllegalStateException {
+
+ // check if a PROPERTY_CONTROLLER_NAME property is present
+ // if so, override the given name
+
+ String propertyControllerName = properties.getProperty(PolicyProperties.PROPERTY_CONTROLLER_NAME);
+ if (propertyControllerName != null && !propertyControllerName.isEmpty()) {
+ if (!propertyControllerName.equals(name)) {
+ throw new IllegalStateException("Proposed name (" + name +
+ ") and properties name (" + propertyControllerName +
+ ") don't match");
+ }
+ name = propertyControllerName;
+ }
+
+ // feature hook
+ for (FeatureAPI feature : FeatureAPI.impl.getList()) {
+ feature.beforeCreateController(name, properties);
+ }
+
+ PolicyController controller = PolicyController.factory.build(name, properties);
+ if (this.isLocked())
+ controller.lock();
+
+ // feature hook
+ for (FeatureAPI feature : FeatureAPI.impl.getList()) {
+ // NOTE: this should change to the actual controller object
+ feature.afterCreateController(name);
+ }
+
+ return controller;
+ }
+
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean configure(PdpdConfiguration config) throws IllegalArgumentException, IllegalStateException {
+
+ if (config == null)
+ throw new IllegalArgumentException("No configuration provided");
+
+ String entity = config.getEntity();
+
+ switch (entity) {
+ case PdpdConfiguration.CONFIG_ENTITY_CONTROLLER:
+ /* only this one supported for now */
+ List<ControllerConfiguration> configControllers = config.getControllers();
+ if (configControllers == null || configControllers.isEmpty()) {
+ if (logger.isInfoEnabled())
+ logger.info("No controller configuration provided: " + config);
+ return false;
+ }
+ List<PolicyController> policyControllers = this.updatePolicyControllers(config.getControllers());
+ if (policyControllers == null || policyControllers.isEmpty())
+ return false;
+ else if (policyControllers.size() == configControllers.size())
+ return true;
+
+ return false;
+ default:
+ String msg = "Configuration Entity is not supported: " + entity;
+ logger.warn(msg);
+ throw new IllegalArgumentException(msg);
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public List<PolicyController> updatePolicyControllers(List<ControllerConfiguration> configControllers)
+ throws IllegalArgumentException, IllegalStateException {
+
+ List<PolicyController> policyControllers = new ArrayList<PolicyController>();
+ if (configControllers == null || configControllers.isEmpty()) {
+ if (logger.isInfoEnabled())
+ logger.info("No controller configuration provided: " + configControllers);
+ return policyControllers;
+ }
+
+ for (ControllerConfiguration configController: configControllers) {
+ try {
+ PolicyController policyController = this.updatePolicyController(configController);
+ policyControllers.add(policyController);
+ } catch (Exception e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, "PolicyEngine", "updatePolicyControllers");
+ }
+ }
+
+ return policyControllers;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public PolicyController updatePolicyController(ControllerConfiguration configController)
+ throws Exception {
+
+ if (configController == null)
+ throw new IllegalArgumentException("No controller configuration has been provided");
+
+ String controllerName = configController.getName();
+ if (controllerName == null || controllerName.isEmpty()) {
+ logger.warn("controller-name must be provided");
+ throw new IllegalArgumentException("No controller configuration has been provided");
+ }
+
+ PolicyController policyController = null;
+ try {
+ String operation = configController.getOperation();
+ if (operation == null || operation.isEmpty()) {
+ logger.warn("operation must be provided");
+ throw new IllegalArgumentException("operation must be provided");
+ }
+
+ try {
+ policyController = PolicyController.factory.get(controllerName);
+ } catch (IllegalArgumentException e) {
+ // not found
+ logger.warn("Policy Controller " + controllerName + " not found");
+ }
+
+ if (policyController == null) {
+
+ if (operation.equalsIgnoreCase(ControllerConfiguration.CONFIG_CONTROLLER_OPERATION_LOCK) ||
+ operation.equalsIgnoreCase(ControllerConfiguration.CONFIG_CONTROLLER_OPERATION_UNLOCK)) {
+ throw new IllegalArgumentException(controllerName + " is not available for operation " + operation);
+ }
+
+ /* Recovery case */
+
+ logger.warn("controller " + controllerName + " does not exist. " +
+ "Attempting recovery from disk");
+
+ Properties properties =
+ SystemPersistence.manager.getControllerProperties(controllerName);
+
+ /*
+ * returned properties cannot be null (per implementation)
+ * assert (properties != null)
+ */
+
+ if (properties == null) {
+ throw new IllegalArgumentException(controllerName + " is invalid");
+ }
+
+ logger.warn("controller " + controllerName + " being recovered. " +
+ "Reset controller's bad maven coordinates to brainless");
+
+ /*
+ * try to bring up bad controller in brainless mode,
+ * after having it working, apply the new create/update operation.
+ */
+ properties.setProperty(PolicyProperties.RULES_GROUPID, DroolsController.NO_GROUP_ID);
+ properties.setProperty(PolicyProperties.RULES_ARTIFACTID, DroolsController.NO_ARTIFACT_ID);
+ properties.setProperty(PolicyProperties.RULES_VERSION, DroolsController.NO_VERSION);
+
+ policyController = PolicyEngine.manager.createPolicyController(controllerName, properties);
+
+ /* fall through to do brain update operation*/
+ }
+
+ switch (operation) {
+ case ControllerConfiguration.CONFIG_CONTROLLER_OPERATION_CREATE:
+ PolicyController.factory.patch(policyController, configController.getDrools());
+ break;
+ case ControllerConfiguration.CONFIG_CONTROLLER_OPERATION_UPDATE:
+ policyController.unlock();
+ PolicyController.factory.patch(policyController, configController.getDrools());
+ break;
+ case ControllerConfiguration.CONFIG_CONTROLLER_OPERATION_LOCK:
+ policyController.lock();
+ break;
+ case ControllerConfiguration.CONFIG_CONTROLLER_OPERATION_UNLOCK:
+ policyController.unlock();
+ break;
+ default:
+ String msg = "Controller Operation Configuration is not supported: " +
+ operation + " for " + controllerName;
+ logger.warn(msg);
+ throw new IllegalArgumentException(msg);
+ }
+
+ return policyController;
+ } catch (Exception e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, "PolicyEngine", "updatePolicyController " + e.getMessage());
+ throw e;
+ } catch (LinkageError e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, "PolicyEngine", "updatePolicyController " + e.getMessage());
+ throw new IllegalStateException(e);
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean start() throws IllegalStateException {
+
+ if (this.locked) {
+ throw new IllegalStateException("Engine is locked");
+ }
+
+ // Features hook
+ for (FeatureAPI feature : FeatureAPI.impl.getList()) {
+ feature.beforeStartEngine();
+ }
+
+ synchronized(this) {
+ this.alive = true;
+ }
+
+ boolean success = true;
+
+ /* Start Policy Engine exclusively-owned (unmanaged) http servers */
+
+ for (HttpServletServer httpServer: this.httpServers) {
+ try {
+ if (!httpServer.start())
+ success = false;
+ } catch (Exception e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, httpServer.toString(), this.toString());
+ }
+ }
+ /* Start Policy Engine exclusively-owned (unmanaged) sources */
+
+ for (TopicSource source: this.sources) {
+ try {
+ if (!source.start())
+ success = false;
+ } catch (Exception e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, source.toString(), this.toString());
+ }
+ }
+
+ /* Start Policy Engine owned (unmanaged) sinks */
+
+ for (TopicSink sink: this.sinks) {
+ try {
+ if (!sink.start())
+ success = false;
+ } catch (Exception e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, sink.toString(), this.toString());
+ }
+ }
+
+ /* Start Policy Controllers */
+
+ List<PolicyController> controllers = PolicyController.factory.inventory();
+ for (PolicyController controller : controllers) {
+ try {
+ if (!controller.start())
+ success = false;
+ } catch (Exception e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, controller.toString(), this.toString());
+ success = false;
+ }
+ }
+
+ /* Start managed Topic Endpoints */
+
+ try {
+ if (!TopicEndpoint.manager.start())
+ success = false;
+ } catch (IllegalStateException e) {
+ String msg = "Topic Endpoint Manager is in an invalid state: " + e.getMessage() + " : " + this;
+ logger.warn(msg);
+ }
+
+
+ // Start the JMX listener
+
+ PdpJmxListener.start();
+
+ // Features hook
+ for (FeatureAPI feature : FeatureAPI.impl.getList()) {
+ feature.afterStartEngine();
+ }
+
+ return success;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean stop() {
+
+ /* stop regardless of the lock state */
+
+ synchronized(this) {
+ if (!this.alive)
+ return true;
+
+ this.alive = false;
+ }
+
+ boolean success = true;
+ List<PolicyController> controllers = PolicyController.factory.inventory();
+ for (PolicyController controller : controllers) {
+ try {
+ if (!controller.stop())
+ success = false;
+ } catch (Exception e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, controller.toString(), this.toString());
+ success = false;
+ }
+ }
+
+ /* Stop Policy Engine owned (unmanaged) sources */
+ for (TopicSource source: this.sources) {
+ try {
+ if (!source.stop())
+ success = false;
+ } catch (Exception e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, source.toString(), this.toString());
+ }
+ }
+
+ /* Stop Policy Engine owned (unmanaged) sinks */
+ for (TopicSink sink: this.sinks) {
+ try {
+ if (!sink.stop())
+ success = false;
+ } catch (Exception e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, sink.toString(), this.toString());
+ }
+ }
+
+ /* stop all managed topics sources and sinks */
+ if (!TopicEndpoint.manager.stop())
+ success = false;
+
+ /* stop all unmanaged http servers */
+ for (HttpServletServer httpServer: this.httpServers) {
+ try {
+ if (!httpServer.stop())
+ success = false;
+ } catch (Exception e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, httpServer.toString(), this.toString());
+ }
+ }
+
+ return success;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void shutdown() throws IllegalStateException {
+
+ synchronized(this) {
+ this.alive = false;
+ }
+
+ // feature hook reporting that the Policy Engine is being shut down
+ for (FeatureAPI feature : FeatureAPI.impl.getList()) {
+ feature.beforeShutdownEngine();
+ }
+
+ /* Shutdown Policy Engine owned (unmanaged) sources */
+ for (TopicSource source: this.sources) {
+ try {
+ source.shutdown();
+ } catch (Exception e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, source.toString(), this.toString());
+ }
+ }
+
+ /* Shutdown Policy Engine owned (unmanaged) sinks */
+ for (TopicSink sink: this.sinks) {
+ try {
+ sink.shutdown();
+ } catch (Exception e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, sink.toString(), this.toString());
+ }
+ }
+
+ /* Shutdown managed resources */
+ PolicyController.factory.shutdown();
+ TopicEndpoint.manager.shutdown();
+ HttpServletServer.factory.destroy();
+
+ // Stop the JMX listener
+
+ PdpJmxListener.stop();
+
+ // feature hook reporting that the Policy Engine has being shut down
+ for (FeatureAPI feature : FeatureAPI.impl.getList()) {
+ feature.afterShutdownEngine();
+ }
+
+ new Thread(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ Thread.sleep(5000L);
+ } catch (InterruptedException e) {
+ logger.warn("InterruptedException while shutting down management server: " + this.toString());
+ }
+
+ /* shutdown all unmanaged http servers */
+ for (HttpServletServer httpServer: getHttpServers()) {
+ try {
+ httpServer.shutdown();
+ } catch (Exception e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, httpServer.toString(), this.toString());
+ }
+ }
+
+ try {
+ Thread.sleep(5000L);
+ } catch (InterruptedException e) {
+ logger.warn("InterruptedException while shutting down management server: " + this.toString());
+ }
+
+ System.exit(0);
+ }
+ }).start();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean isAlive() {
+ return this.alive;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean lock() {
+
+ synchronized(this) {
+ if (this.locked)
+ return true;
+
+ this.locked = true;
+ }
+
+ boolean success = true;
+ List<PolicyController> controllers = PolicyController.factory.inventory();
+ for (PolicyController controller : controllers) {
+ try {
+ success = controller.lock() && success;
+ } catch (Exception e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, controller.toString(), this.toString());
+ success = false;
+ }
+ }
+
+ success = TopicEndpoint.manager.lock();
+ return success;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean unlock() {
+ synchronized(this) {
+ if (!this.locked)
+ return true;
+
+ this.locked = false;
+ }
+
+ boolean success = true;
+ List<PolicyController> controllers = PolicyController.factory.inventory();
+ for (PolicyController controller : controllers) {
+ try {
+ success = controller.unlock() && success;
+ } catch (Exception e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, controller.toString(), this.toString());
+ success = false;
+ }
+ }
+
+ success = TopicEndpoint.manager.unlock();
+ return success;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean isLocked() {
+ return this.locked;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void removePolicyController(String name) {
+ PolicyController.factory.destroy(name);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void removePolicyController(PolicyController controller) {
+ PolicyController.factory.destroy(controller);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @JsonIgnore
+ @Override
+ public List<PolicyController> getPolicyControllers() {
+ return PolicyController.factory.inventory();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public List<String> getControllers() {
+ List<String> controllerNames = new ArrayList<String>();
+ for (PolicyController controller: PolicyController.factory.inventory()) {
+ controllerNames.add(controller.getName());
+ }
+ return controllerNames;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Properties getProperties() {
+ return this.properties;
+ }
+
+
+ /**
+ * {@inheritDoc}
+ */
+ @SuppressWarnings("unchecked")
+ @Override
+ public List<TopicSource> getSources() {
+ return (List<TopicSource>) this.sources;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @SuppressWarnings("unchecked")
+ @Override
+ public List<TopicSink> getSinks() {
+ return (List<TopicSink>) this.sinks;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public List<HttpServletServer> getHttpServers() {
+ return this.httpServers;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean onTopicEvent(CommInfrastructure commType, String topic, String event) {
+ /* configuration request */
+ try {
+ PdpdConfiguration configuration = this.decoder.fromJson(event, PdpdConfiguration.class);
+ this.configure(configuration);
+ } catch (Exception e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, "CONFIGURATION ERROR IN PDP-D POLICY ENGINE: "+ event + ":" + e.getMessage() + ":" + this);
+ }
+
+ return true;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean deliver(String topic, Object event)
+ throws IllegalArgumentException, IllegalStateException {
+
+ /*
+ * Note this entry point is usually from the DRL
+ */
+
+ if (topic == null || topic.isEmpty())
+ throw new IllegalArgumentException("Invalid Topic");
+
+ if (event == null)
+ throw new IllegalArgumentException("Invalid Event");
+
+ if (!this.isAlive())
+ throw new IllegalStateException("Policy Engine is stopped");
+
+ if (this.isLocked())
+ throw new IllegalStateException("Policy Engine is locked");
+
+ List<? extends TopicSink> sinks =
+ TopicEndpoint.manager.getTopicSinks(topic);
+ if (sinks == null || sinks.isEmpty() || sinks.size() > 1)
+ throw new IllegalStateException
+ ("Cannot ensure correct delivery on topic " + topic + ": " + sinks);
+
+ return this.deliver(sinks.get(0).getTopicCommInfrastructure(),
+ topic, event);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean deliver(String busType, String topic, Object event)
+ throws IllegalArgumentException, IllegalStateException,
+ UnsupportedOperationException {
+
+ /*
+ * Note this entry point is usually from the DRL (one of the reasons
+ * busType is String.
+ */
+
+ if (busType == null || busType.isEmpty())
+ throw new IllegalArgumentException
+ ("Invalid Communication Infrastructure");
+
+ if (topic == null || topic.isEmpty())
+ throw new IllegalArgumentException("Invalid Topic");
+
+ if (event == null)
+ throw new IllegalArgumentException("Invalid Event");
+
+ boolean valid = false;
+ for (Topic.CommInfrastructure comm: Topic.CommInfrastructure.values()) {
+ if (comm.name().equals(busType)) {
+ valid = true;
+ }
+ }
+
+ if (!valid)
+ throw new IllegalArgumentException
+ ("Invalid Communication Infrastructure: " + busType);
+
+
+ if (!this.isAlive())
+ throw new IllegalStateException("Policy Engine is stopped");
+
+ if (this.isLocked())
+ throw new IllegalStateException("Policy Engine is locked");
+
+
+ return this.deliver(Topic.CommInfrastructure.valueOf(busType),
+ topic, event);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean deliver(Topic.CommInfrastructure busType,
+ String topic, Object event)
+ throws IllegalArgumentException, IllegalStateException,
+ UnsupportedOperationException {
+
+ if (topic == null || topic.isEmpty())
+ throw new IllegalArgumentException("Invalid Topic");
+
+ if (event == null)
+ throw new IllegalArgumentException("Invalid Event");
+
+ if (!this.isAlive())
+ throw new IllegalStateException("Policy Engine is stopped");
+
+ if (this.isLocked())
+ throw new IllegalStateException("Policy Engine is locked");
+
+ /* Try to send through the controller, this is the
+ * preferred way, since it may want to apply additional
+ * processing
+ */
+ try {
+ DroolsController droolsController =
+ EventProtocolCoder.manager.getDroolsController(topic, event);
+ PolicyController controller = PolicyController.factory.get(droolsController);
+ if (controller != null)
+ return controller.deliver(busType, topic, event);
+ } catch (Exception e) {
+ logger.warn(MessageCodes.EXCEPTION_ERROR, e,
+ busType + ":" + topic + " :" + event, this.toString());
+ /* continue (try without routing through the controller) */
+ }
+
+ /*
+ * cannot route through the controller, send directly through
+ * the topic sink
+ */
+ try {
+ String json = EventProtocolCoder.manager.encode(topic, event);
+ return this.deliver(busType, topic, json);
+
+ } catch (Exception e) {
+ logger.warn(MessageCodes.EXCEPTION_ERROR, e,
+ busType + ":" + topic + " :" + event, this.toString());
+ throw e;
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean deliver(Topic.CommInfrastructure busType,
+ String topic, String event)
+ throws IllegalArgumentException, IllegalStateException,
+ UnsupportedOperationException {
+
+ if (topic == null || topic.isEmpty())
+ throw new IllegalArgumentException("Invalid Topic");
+
+ if (event == null || event.isEmpty())
+ throw new IllegalArgumentException("Invalid Event");
+
+ if (!this.isAlive())
+ throw new IllegalStateException("Policy Engine is stopped");
+
+ if (this.isLocked())
+ throw new IllegalStateException("Policy Engine is locked");
+
+ try {
+ TopicSink sink =
+ TopicEndpoint.manager.getTopicSink
+ (busType, topic);
+
+ if (sink == null)
+ throw new IllegalStateException("Inconsistent State: " + this);
+
+ return sink.send(event);
+
+ } catch (Exception e) {
+ logger.warn(MessageCodes.EXCEPTION_ERROR, e,
+ busType + ":" + topic + " :" + event, this.toString());
+ throw e;
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public synchronized void activate() {
+
+ // activate 'policy-management'
+ for (PolicyController policyController : getPolicyControllers()) {
+ try {
+ policyController.unlock();
+ policyController.start();
+ } catch (Exception e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, "PolicyEngine.activate: cannot start " +
+ policyController + " because of " + e.getMessage());
+ } catch (LinkageError e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, "PolicyEngine.activate: cannot start " +
+ policyController + " because of " + e.getMessage());
+ }
+ }
+
+ this.unlock();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public synchronized void deactivate() {
+
+ this.lock();
+
+ for (PolicyController policyController : getPolicyControllers()) {
+ try {
+ policyController.stop();
+ } catch (Exception e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, "PolicyEngine.deactivate: cannot stop " +
+ policyController + " because of " + e.getMessage());
+ } catch (LinkageError e) {
+ logger.error(MessageCodes.EXCEPTION_ERROR, e, "PolicyEngine.deactivate: cannot start " +
+ policyController + " because of " + e.getMessage());
+ }
+ }
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder builder = new StringBuilder();
+ builder.append("PolicyEngineManager [alive=").append(alive).append(", locked=").append(locked).append("]");
+ return builder.toString();
+ }
+
+}
+
+