diff options
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.java | 1182 |
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(); + } + +} + + |