package org.onap.policy.pdpx.main.startstop;
+import java.util.Arrays;
+import java.util.Properties;
+import org.onap.policy.common.endpoints.event.comm.TopicEndpoint;
+import org.onap.policy.common.endpoints.event.comm.TopicSource;
+import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClient;
+import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClientException;
+import org.onap.policy.common.endpoints.listeners.MessageTypeDispatcher;
import org.onap.policy.common.parameters.ParameterService;
+import org.onap.policy.common.utils.services.ServiceManagerContainer;
+import org.onap.policy.models.pdp.concepts.PdpStatus;
+import org.onap.policy.models.pdp.concepts.PdpUpdate;
+import org.onap.policy.models.pdp.enums.PdpMessageType;
+import org.onap.policy.models.pdp.enums.PdpState;
import org.onap.policy.pdpx.main.PolicyXacmlPdpException;
+import org.onap.policy.pdpx.main.PolicyXacmlPdpRuntimeException;
+import org.onap.policy.pdpx.main.comm.XacmlPdpMessage;
+import org.onap.policy.pdpx.main.comm.XacmlPdpPapRegistration;
+import org.onap.policy.pdpx.main.comm.listeners.XacmlPdpStateChangeListener;
import org.onap.policy.pdpx.main.parameters.XacmlPdpParameterGroup;
import org.onap.policy.pdpx.main.rest.XacmlPdpRestServer;
import org.slf4j.Logger;
* This class wraps a distributor so that it can be activated as a complete service together with
* all its xacml pdp and forwarding handlers.
*/
-public class XacmlPdpActivator {
+public class XacmlPdpActivator extends ServiceManagerContainer {
+
// The logger for this class
private static final Logger LOGGER = LoggerFactory.getLogger(XacmlPdpActivator.class);
+ private static final String[] MSG_TYPE_NAMES = {"messageName"};
+ private static final String TOPIC = "POLICY-PDP-PAP";
+
// The parameters of this policy xacml pdp activator
private final XacmlPdpParameterGroup xacmlPdpParameterGroup;
- private static boolean alive = false;
-
+ /**
+ * The XACML PDP REST API server.
+ */
private XacmlPdpRestServer restServer;
/**
- * Instantiate the activator for policy xacml pdp as a complete service.
- *
- * @param xacmlPdpParameterGroup the parameters for the xacml pdp service
+ * Listens for messages on the topic, decodes them into a {@link PdpStatus} message, and then
+ * dispatches them to {@link #pdpUpdateListener}.
*/
- public XacmlPdpActivator(final XacmlPdpParameterGroup xacmlPdpParameterGroup) {
- this.xacmlPdpParameterGroup = xacmlPdpParameterGroup;
- }
+ private final MessageTypeDispatcher msgDispatcher;
/**
- * Initialize xacml pdp as a complete service.
- *
- * @throws PolicyXacmlPdpException on errors in initializing the service
+ * Listens for {@link PdpUpdate} messages and then routes them to the listener associated with the
+ * ID of the originating request.
*/
- public void initialize() throws PolicyXacmlPdpException {
- LOGGER.debug("Policy xacml pdp starting as a service . . .");
- startXacmlPdpRestServer();
- registerToParameterService(xacmlPdpParameterGroup);
- XacmlPdpActivator.setAlive(true);
- LOGGER.debug("Policy xacml pdp started as a service");
- }
+ private final XacmlPdpStateChangeListener pdpStateChangeListener;
/**
- * Starts the xacml pdp rest server using configuration parameters.
- *
- * @throws PolicyXacmlPdpException if server start fails
+ * The current activator.
*/
- private void startXacmlPdpRestServer() throws PolicyXacmlPdpException {
- xacmlPdpParameterGroup.getRestServerParameters().setName(xacmlPdpParameterGroup.getName());
- restServer = new XacmlPdpRestServer(xacmlPdpParameterGroup.getRestServerParameters(),
- xacmlPdpParameterGroup.getApplicationPath());
- if (!restServer.start()) {
- throw new PolicyXacmlPdpException("Failed to start xacml pdp rest server. Check log for more details...");
- }
- }
+ private static XacmlPdpActivator current = null;
+
+ private volatile boolean alive = false;
+
+ private XacmlPdpPapRegistration register;
+
+ private XacmlPdpMessage message;
/**
- * Terminate policy xacml pdp.
+ * Instantiate the activator for policy xacml pdp as a complete service.
*
- * @throws PolicyXacmlPdpException on termination errors
+ * @param xacmlPdpParameterGroup the parameters for the xacml pdp service
+ * @param topicProperties properties used to configure the topics
*/
- public void terminate() throws PolicyXacmlPdpException {
+ public XacmlPdpActivator(final XacmlPdpParameterGroup xacmlPdpParameterGroup, Properties topicProperties) {
+ TopicEndpoint.manager.addTopicSinks(topicProperties);
+ TopicEndpoint.manager.addTopicSources(topicProperties);
+
try {
- deregisterToParameterService(xacmlPdpParameterGroup);
- XacmlPdpActivator.setAlive(false);
-
- // Stop the xacml pdp rest server
- restServer.stop();
- } catch (final Exception exp) {
- LOGGER.error("Policy xacml pdp service termination failed", exp);
- throw new PolicyXacmlPdpException(exp.getMessage(), exp);
+ TopicSinkClient sinkClient = new TopicSinkClient(TOPIC);
+ this.xacmlPdpParameterGroup = xacmlPdpParameterGroup;
+ this.msgDispatcher = new MessageTypeDispatcher(MSG_TYPE_NAMES);
+ this.pdpStateChangeListener = new XacmlPdpStateChangeListener(sinkClient);
+ this.register = new XacmlPdpPapRegistration(sinkClient);
+ this.message = new XacmlPdpMessage();
+ } catch (RuntimeException | TopicSinkClientException e) {
+ throw new PolicyXacmlPdpRuntimeException(e.getMessage(), e);
}
+
+ xacmlPdpParameterGroup.getRestServerParameters().setName(xacmlPdpParameterGroup.getName());
+
+ // @formatter:off
+ addAction("XACML PDP parameters", () -> ParameterService.register(xacmlPdpParameterGroup),
+ () -> ParameterService.deregister(xacmlPdpParameterGroup.getName()));
+
+ addAction("Request ID Dispatcher",
+ () -> msgDispatcher.register(PdpMessageType.PDP_STATE_CHANGE.name(), this.pdpStateChangeListener),
+ () -> msgDispatcher.unregister(PdpMessageType.PDP_STATE_CHANGE.name()));
+
+ addAction("Message Dispatcher",
+ () -> registerMsgDispatcher(),
+ () -> unregisterMsgDispatcher());
+
+ addAction("topics",
+ () -> TopicEndpoint.manager.start(),
+ () -> TopicEndpoint.manager.shutdown());
+
+ addAction("Create REST server",
+ () -> {
+ restServer = new XacmlPdpRestServer(xacmlPdpParameterGroup.getRestServerParameters(),
+ xacmlPdpParameterGroup.getApplicationPath());
+ },
+ () -> {
+ restServer = null;
+ });
+
+ addAction("REST server",
+ () -> restServer.start(),
+ () -> restServer.stop());
+
+ addAction("set alive", () -> setAlive(true), () -> setAlive(false));
+
+ addAction("Initial Registration with PAP",
+ () -> {
+ register.pdpRegistration(message.formatStatusMessage(PdpState.PASSIVE));
+ },
+ () -> {
+ register.pdpRegistration(message.formatStatusMessage(PdpState.TERMINATED));
+ });
+ // @formatter:on
+
+ current = this;
}
/**
ParameterService.deregister(xacmlPdpParameterGroup.getName());
}
+ /**
+ * Registers the dispatcher with the topic source(s).
+ */
+ private void registerMsgDispatcher() {
+ for (TopicSource source : TopicEndpoint.manager.getTopicSources(Arrays.asList(TOPIC))) {
+ source.register(msgDispatcher);
+ }
+ }
+
+ /**
+ * Unregisters the dispatcher from the topic source(s).
+ */
+ private void unregisterMsgDispatcher() {
+ for (TopicSource source : TopicEndpoint.manager.getTopicSources(Arrays.asList(TOPIC))) {
+ source.unregister(msgDispatcher);
+ }
+ }
+
/**
* Returns the alive status of xacml pdp service.
*
* @return the alive
*/
- public static boolean isAlive() {
+ @Override
+ public boolean isAlive() {
return alive;
}
*
* @param status the status
*/
- private static void setAlive(final boolean status) {
+ private void setAlive(final boolean status) {
alive = status;
}
+
+ public static XacmlPdpActivator getCurrent() {
+ return current;
+ }
+
+ public static void setCurrent(XacmlPdpActivator current) {
+ XacmlPdpActivator.current = current;
+ }
}
Code Review / policy / xacml-pdp.git / blobdiff