Merge "Adding plugins-event module to apex-pdp"

5 files changed, 539 insertions, 0 deletions

diff --git a/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-websocket/pom.xml b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-websocket/pom.xml
new file mode 100644
index 000000000..f6068aa28
--- /dev/null
+++ b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-websocket/pom.xml
@@ -0,0 +1,36 @@
+<!--
+  ============LICENSE_START=======================================================
+   Copyright (C) 2018 Ericsson. 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.
+
+  SPDX-License-Identifier: Apache-2.0
+  ============LICENSE_END=========================================================
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.onap.policy.apex-pdp.plugins.plugins-event.plugins-event-carrier</groupId>
+        <artifactId>plugins-event-carrier</artifactId>
+        <version>2.0.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>plugins-event-carrier-websocket</artifactId>
+    <name>${project.artifactId}</name>
+    <description>[${project.parent.artifactId}] Plugin for handling events being transported over web sockets</description>
+
+    <properties>
+        <apex-plugins-event-carrier-websocket-dir>${project.basedir}/src</apex-plugins-event-carrier-websocket-dir>
+    </properties>
+</project>
\ No newline at end of file
diff --git a/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-websocket/src/main/java/org/onap/policy/apex/plugins/event/carrier/websocket/ApexWebSocketConsumer.java b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-websocket/src/main/java/org/onap/policy/apex/plugins/event/carrier/websocket/ApexWebSocketConsumer.java
new file mode 100644
index 000000000..f2c59453c
--- /dev/null
+++ b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-websocket/src/main/java/org/onap/policy/apex/plugins/event/carrier/websocket/ApexWebSocketConsumer.java
@@ -0,0 +1,195 @@
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2016-2018 Ericsson. 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.apex.plugins.event.carrier.websocket;
+
+import java.util.EnumMap;
+import java.util.Map;
+
+import org.onap.policy.apex.core.infrastructure.messaging.MessagingException;
+import org.onap.policy.apex.core.infrastructure.messaging.stringmessaging.WSStringMessageClient;
+import org.onap.policy.apex.core.infrastructure.messaging.stringmessaging.WSStringMessageListener;
+import org.onap.policy.apex.core.infrastructure.messaging.stringmessaging.WSStringMessageServer;
+import org.onap.policy.apex.core.infrastructure.messaging.stringmessaging.WSStringMessager;
+import org.onap.policy.apex.core.infrastructure.threading.ApplicationThreadFactory;
+import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities;
+import org.onap.policy.apex.service.engine.event.ApexEventConsumer;
+import org.onap.policy.apex.service.engine.event.ApexEventException;
+import org.onap.policy.apex.service.engine.event.ApexEventReceiver;
+import org.onap.policy.apex.service.engine.event.PeeredReference;
+import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerParameters;
+import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerPeeredMode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Concrete implementation an Apex event consumer that receives events using Kafka.
+ *
+ * @author Liam Fallon (liam.fallon@ericsson.com)
+ */
+public class ApexWebSocketConsumer implements ApexEventConsumer, WSStringMessageListener, Runnable {
+    private static final int WEB_SOCKET_WAIT_SLEEP_TIME = 100;
+
+    // Get a reference to the logger
+    private static final Logger LOGGER = LoggerFactory.getLogger(ApexWebSocketConsumer.class);
+
+    // The Web Socket properties
+    private WEBSOCKETCarrierTechnologyParameters webSocketConsumerProperties;
+
+    // The web socket messager, may be WS a server or a client
+    private WSStringMessager wsStringMessager;
+
+    // The event receiver that will receive events from this consumer
+    private ApexEventReceiver eventReceiver;
+
+    // The name for this consumer
+    private String name = null;
+
+    // The peer references for this event handler
+    private Map<EventHandlerPeeredMode, PeeredReference> peerReferenceMap = new EnumMap<>(EventHandlerPeeredMode.class);
+
+    // The consumer thread and stopping flag
+    private Thread consumerThread;
+    private boolean stopOrderedFlag = false;
+
+    // The number of events read to date
+    private int eventsRead = 0;
+
+    @Override
+    public void init(final String consumerName, final EventHandlerParameters consumerParameters,
+            final ApexEventReceiver incomingEventReceiver) throws ApexEventException {
+        this.eventReceiver = incomingEventReceiver;
+        this.name = consumerName;
+
+        // Check and get the Kafka Properties
+        if (!(consumerParameters.getCarrierTechnologyParameters() instanceof WEBSOCKETCarrierTechnologyParameters)) {
+            LOGGER.warn("specified consumer properties are not applicable to a web socket consumer");
+            throw new ApexEventException("specified consumer properties are not applicable to a web socket consumer");
+        }
+        webSocketConsumerProperties =
+                (WEBSOCKETCarrierTechnologyParameters) consumerParameters.getCarrierTechnologyParameters();
+
+        // Check if this is a server or a client Web Socket
+        if (webSocketConsumerProperties.isWsClient()) {
+            // Create a WS client
+            wsStringMessager = new WSStringMessageClient(webSocketConsumerProperties.getHost(),
+                    webSocketConsumerProperties.getPort());
+        } else {
+            wsStringMessager = new WSStringMessageServer(webSocketConsumerProperties.getPort());
+        }
+
+        // Start reception of event strings on the web socket
+        try {
+            wsStringMessager.start(this);
+        } catch (final MessagingException e) {
+            LOGGER.warn("could not start web socket consumer");
+        }
+    }
+
+    /*
+     * (non-Javadoc)
+     *
+     * @see org.onap.policy.apex.service.engine.event.ApexEventConsumer#start()
+     */
+    @Override
+    public void start() {
+        // Configure and start the event reception thread
+        final String threadName = this.getClass().getName() + ":" + this.name;
+        consumerThread = new ApplicationThreadFactory(threadName).newThread(this);
+        consumerThread.setDaemon(true);
+        consumerThread.start();
+    }
+
+    /*
+     * (non-Javadoc)
+     *
+     * @see org.onap.policy.apex.service.engine.event.ApexEventConsumer#getName()
+     */
+    @Override
+    public String getName() {
+        return name;
+    }
+
+    /*
+     * (non-Javadoc)
+     *
+     * @see org.onap.policy.apex.service.engine.event.ApexEventConsumer#getPeeredReference(org.onap.policy.apex.service.
+     * parameters. eventhandler.EventHandlerPeeredMode)
+     */
+    @Override
+    public PeeredReference getPeeredReference(final EventHandlerPeeredMode peeredMode) {
+        return peerReferenceMap.get(peeredMode);
+    }
+
+    /*
+     * (non-Javadoc)
+     *
+     * @see org.onap.policy.apex.service.engine.event.ApexEventConsumer#setPeeredReference(org.onap.policy.apex.service.
+     * parameters. eventhandler.EventHandlerPeeredMode, org.onap.policy.apex.service.engine.event.PeeredReference)
+     */
+    @Override
+    public void setPeeredReference(final EventHandlerPeeredMode peeredMode, final PeeredReference peeredReference) {
+        peerReferenceMap.put(peeredMode, peeredReference);
+    }
+
+    /*
+     * (non-Javadoc)
+     *
+     * @see java.lang.Runnable#run()
+     */
+    @Override
+    public void run() {
+        while (consumerThread.isAlive() && !stopOrderedFlag) {
+            ThreadUtilities.sleep(WEB_SOCKET_WAIT_SLEEP_TIME);
+        }
+    }
+
+    /*
+     * (non-Javadoc)
+     *
+     * @see org.onap.policy.apex.apps.uservice.producer.ApexEventProducer#stop()
+     */
+    @Override
+    public void stop() {
+        if (wsStringMessager != null) {
+            wsStringMessager.stop();
+        }
+        stopOrderedFlag = true;
+    }
+
+    /*
+     * (non-Javadoc)
+     *
+     * @see
+     * org.onap.policy.apex.core.infrastructure.messaging.stringmessaging.WSStringMessageListener#receiveString(java.
+     * lang. String)
+     */
+    @Override
+    public void receiveString(final String eventString) {
+        try {
+            eventReceiver.receiveEvent(eventString);
+            eventsRead++;
+        } catch (final Exception e) {
+            final String errorMessage = "Error sending event " + name + '_' + eventsRead + ", " + e.getMessage()
+                    + ", event:\n" + eventString;
+            LOGGER.warn(errorMessage);
+        }
+    }
+}
diff --git a/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-websocket/src/main/java/org/onap/policy/apex/plugins/event/carrier/websocket/ApexWebSocketProducer.java b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-websocket/src/main/java/org/onap/policy/apex/plugins/event/carrier/websocket/ApexWebSocketProducer.java
new file mode 100644
index 000000000..1fd146824
--- /dev/null
+++ b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-websocket/src/main/java/org/onap/policy/apex/plugins/event/carrier/websocket/ApexWebSocketProducer.java
@@ -0,0 +1,166 @@
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2016-2018 Ericsson. 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.apex.plugins.event.carrier.websocket;
+
+import java.util.EnumMap;
+import java.util.Map;
+
+import org.onap.policy.apex.core.infrastructure.messaging.MessagingException;
+import org.onap.policy.apex.core.infrastructure.messaging.stringmessaging.WSStringMessageClient;
+import org.onap.policy.apex.core.infrastructure.messaging.stringmessaging.WSStringMessageListener;
+import org.onap.policy.apex.core.infrastructure.messaging.stringmessaging.WSStringMessageServer;
+import org.onap.policy.apex.core.infrastructure.messaging.stringmessaging.WSStringMessager;
+import org.onap.policy.apex.service.engine.event.ApexEventException;
+import org.onap.policy.apex.service.engine.event.ApexEventProducer;
+import org.onap.policy.apex.service.engine.event.PeeredReference;
+import org.onap.policy.apex.service.engine.event.SynchronousEventCache;
+import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerParameters;
+import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerPeeredMode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Concrete implementation of an Apex event producer that sends events using a web socket.
+ *
+ * @author Liam Fallon (liam.fallon@ericsson.com)
+ */
+public class ApexWebSocketProducer implements ApexEventProducer, WSStringMessageListener {
+    // Get a reference to the logger
+    private static final Logger LOGGER = LoggerFactory.getLogger(ApexWebSocketProducer.class);
+
+    // The Web Socket properties
+    private WEBSOCKETCarrierTechnologyParameters webSocketProducerProperties;
+
+    // The web socket messager, may be WS a server or a client
+    private WSStringMessager wsStringMessager;
+
+    // The name for this producer
+    private String name = null;
+
+    // The peer references for this event handler
+    private Map<EventHandlerPeeredMode, PeeredReference> peerReferenceMap = new EnumMap<>(EventHandlerPeeredMode.class);
+
+    @Override
+    public void init(final String producerName, final EventHandlerParameters producerParameters)
+            throws ApexEventException {
+        this.name = producerName;
+
+        // Check and get the web socket Properties
+        if (!(producerParameters.getCarrierTechnologyParameters() instanceof WEBSOCKETCarrierTechnologyParameters)) {
+            LOGGER.warn(
+                    "specified producer properties for " + this.name + "are not applicable to a web socket producer");
+            throw new ApexEventException("specified producer properties are not applicable to a web socket producer");
+        }
+        webSocketProducerProperties =
+                (WEBSOCKETCarrierTechnologyParameters) producerParameters.getCarrierTechnologyParameters();
+
+        // Check if this is a server or a client Web Socket
+        if (webSocketProducerProperties.isWsClient()) {
+            // Create a WS client
+            wsStringMessager = new WSStringMessageClient(webSocketProducerProperties.getHost(),
+                    webSocketProducerProperties.getPort());
+        } else {
+            wsStringMessager = new WSStringMessageServer(webSocketProducerProperties.getPort());
+        }
+
+        // Start reception of event strings on the web socket
+        try {
+            wsStringMessager.start(this);
+        } catch (final MessagingException e) {
+            LOGGER.warn("could not start web socket producer (" + this.name + ")");
+        }
+    }
+
+    /*
+     * (non-Javadoc)
+     *
+     * @see org.onap.policy.apex.service.engine.event.ApexEventProducer#getName()
+     */
+    @Override
+    public String getName() {
+        return name;
+    }
+
+    /*
+     * (non-Javadoc)
+     *
+     * @see org.onap.policy.apex.service.engine.event.ApexEventProducer#getPeeredReference(org.onap.policy.apex.service.
+     * parameters. eventhandler.EventHandlerPeeredMode)
+     */
+    @Override
+    public PeeredReference getPeeredReference(final EventHandlerPeeredMode peeredMode) {
+        return peerReferenceMap.get(peeredMode);
+    }
+
+    /*
+     * (non-Javadoc)
+     *
+     * @see org.onap.policy.apex.service.engine.event.ApexEventProducer#setPeeredReference(org.onap.policy.apex.service.
+     * parameters. eventhandler.EventHandlerPeeredMode, org.onap.policy.apex.service.engine.event.PeeredReference)
+     */
+    @Override
+    public void setPeeredReference(final EventHandlerPeeredMode peeredMode, final PeeredReference peeredReference) {
+        peerReferenceMap.put(peeredMode, peeredReference);
+    }
+
+    /*
+     * (non-Javadoc)
+     *
+     * @see org.onap.policy.apex.apps.uservice.producer.ApexEventProducer#sendEvent(long, java.lang.String,
+     * java.lang.Object)
+     */
+    @Override
+    public void sendEvent(final long executionId, final String eventName, final Object event) {
+        // Check if this is a synchronized event, if so we have received a reply
+        final SynchronousEventCache synchronousEventCache =
+                (SynchronousEventCache) peerReferenceMap.get(EventHandlerPeeredMode.SYNCHRONOUS);
+        if (synchronousEventCache != null) {
+            synchronousEventCache.removeCachedEventToApexIfExists(executionId);
+        }
+
+        wsStringMessager.sendString((String) event);
+    }
+
+    /*
+     * (non-Javadoc)
+     *
+     * @see org.onap.policy.apex.apps.uservice.producer.ApexEventProducer#stop()
+     */
+    @Override
+    public void stop() {
+        if (wsStringMessager != null) {
+            wsStringMessager.stop();
+        }
+    }
+
+    /*
+     * (non-Javadoc)
+     *
+     * @see
+     * org.onap.policy.apex.core.infrastructure.messaging.stringmessaging.WSStringMessageListener#receiveString(java.
+     * lang. String)
+     */
+    @Override
+    public void receiveString(final String messageString) {
+        LOGGER.warn("received message \"" + messageString + "\" on web socket producer (" + this.name
+                + ") , no messages should be received on a web socket producer");
+    }
+}
diff --git a/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-websocket/src/main/java/org/onap/policy/apex/plugins/event/carrier/websocket/WEBSOCKETCarrierTechnologyParameters.java b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-websocket/src/main/java/org/onap/policy/apex/plugins/event/carrier/websocket/WEBSOCKETCarrierTechnologyParameters.java
new file mode 100644
index 000000000..e8009a5d8
--- /dev/null
+++ b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-websocket/src/main/java/org/onap/policy/apex/plugins/event/carrier/websocket/WEBSOCKETCarrierTechnologyParameters.java
@@ -0,0 +1,115 @@
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2016-2018 Ericsson. 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.apex.plugins.event.carrier.websocket;
+
+import org.onap.policy.apex.service.parameters.carriertechnology.CarrierTechnologyParameters;
+
+/**
+ * Apex parameters for Kafka as an event carrier technology.
+ *
+ * @author Liam Fallon (liam.fallon@ericsson.com)
+ */
+public class WEBSOCKETCarrierTechnologyParameters extends CarrierTechnologyParameters {
+    // @formatter:off
+    private static final int MIN_USER_PORT =  1024;
+    private static final int MAX_USER_PORT = 65535;
+
+    /** The label of this carrier technology. */
+    public static final String WEB_SCOKET_CARRIER_TECHNOLOGY_LABEL = "WEBSOCKET";
+
+    /** The producer plugin class for the web socket carrier technology. */
+    public static final String WEB_SCOKET_EVENT_PRODUCER_PLUGIN_CLASS = ApexWebSocketProducer.class.getCanonicalName();
+
+    /** The consumer plugin class for the web socket carrier technology. */
+    public static final String KWEB_SCOKET_EVENT_CONSUMER_PLUGIN_CLASS = ApexWebSocketConsumer.class.getCanonicalName();
+
+    // Default parameter values
+    private static final String DEFAULT_HOST = "localhost";
+    private static final int    DEFAULT_PORT = -1;
+
+    // Web socket parameters
+    private boolean wsClient = true;
+    private String  host     = DEFAULT_HOST;
+    private int     port     = DEFAULT_PORT;
+    // @formatter:on
+
+    /**
+     * Constructor to create a web socket carrier technology parameters instance and register the instance with the
+     * parameter service.
+     */
+    public WEBSOCKETCarrierTechnologyParameters() {
+        super(WEBSOCKETCarrierTechnologyParameters.class.getCanonicalName());
+
+        // Set the carrier technology properties for the web socket carrier technology
+        this.setLabel(WEB_SCOKET_CARRIER_TECHNOLOGY_LABEL);
+        this.setEventProducerPluginClass(WEB_SCOKET_EVENT_PRODUCER_PLUGIN_CLASS);
+        this.setEventConsumerPluginClass(KWEB_SCOKET_EVENT_CONSUMER_PLUGIN_CLASS);
+    }
+
+    /**
+     * Gets the host.
+     *
+     * @return the host
+     */
+    public String getHost() {
+        return host;
+    }
+
+    /**
+     * Gets the port.
+     *
+     * @return the port
+     */
+    public int getPort() {
+        return port;
+    }
+
+    /**
+     * Checks if is ws client.
+     *
+     * @return true, if checks if is ws client
+     */
+    public boolean isWsClient() {
+        return wsClient;
+    }
+
+    /*
+     * (non-Javadoc)
+     *
+     * @see org.onap.policy.apex.apps.uservice.parameters.ApexParameterValidator#validate()
+     */
+    @Override
+    public String validate() {
+        final StringBuilder errorMessageBuilder = new StringBuilder();
+
+        errorMessageBuilder.append(super.validate());
+
+        if (wsClient && (host == null || host.trim().length() == 0)) {
+            errorMessageBuilder.append("  host not specified, must be host as a string\n");
+        }
+
+        if (port < MIN_USER_PORT || port > MAX_USER_PORT) {
+            errorMessageBuilder.append("  port [" + port + "] invalid, must be specified as 1024 <= port <= 6535\n");
+        }
+
+        return errorMessageBuilder.toString();
+    }
+}
diff --git a/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-websocket/src/main/java/org/onap/policy/apex/plugins/event/carrier/websocket/package-info.java b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-websocket/src/main/java/org/onap/policy/apex/plugins/event/carrier/websocket/package-info.java
new file mode 100644
index 000000000..9555ea447
--- /dev/null
+++ b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-websocket/src/main/java/org/onap/policy/apex/plugins/event/carrier/websocket/package-info.java
@@ -0,0 +1,27 @@
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2016-2018 Ericsson. 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+/**
+ * Implements the APEX event carrier technology plugin for web sockets.
+ *
+ * @author Liam Fallon (liam.fallon@ericsson.com)
+ */
+
+package org.onap.policy.apex.plugins.event.carrier.websocket;