summaryrefslogtreecommitdiffstats
path: root/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main
diff options
context:
space:
mode:
Diffstat (limited to 'plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main')
-rw-r--r--plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/ApexRestRequest.java65
-rw-r--r--plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/ApexRestRequestorConsumer.java416
-rw-r--r--plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/ApexRestRequestorProducer.java195
-rw-r--r--plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/RESTRequestorCarrierTechnologyParameters.java135
-rw-r--r--plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/package-info.java28
5 files changed, 839 insertions, 0 deletions
diff --git a/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/ApexRestRequest.java b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/ApexRestRequest.java
new file mode 100644
index 000000000..46e68acda
--- /dev/null
+++ b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/ApexRestRequest.java
@@ -0,0 +1,65 @@
+/*-
+ * ============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.restrequestor;
+
+/**
+ * This class holds a record of a REST request for the REST requestor plugin
+ *
+ * @author Liam Fallon (liam.fallon@ericsson.com)
+ */
+public class ApexRestRequest {
+ private long executionId;
+ private String eventName;
+ private Object event;
+ private long timestamp;
+
+ public ApexRestRequest(final long executionId, final String eventName, final Object event) {
+ this.executionId = executionId;
+ this.eventName = eventName;
+ this.event = event;
+ }
+
+ public long getExecutionId() {
+ return executionId;
+ }
+
+ public String getEventName() {
+ return eventName;
+ }
+
+ public Object getEvent() {
+ return event;
+ }
+
+ public long getTimestamp() {
+ return timestamp;
+ }
+
+ public void setTimestamp(final long timestamp) {
+ this.timestamp = timestamp;
+ }
+
+ @Override
+ public String toString() {
+ return "ApexRestRequest [executionId=" + executionId + ", eventName=" + eventName + ", event=" + event
+ + ", timestamp=" + timestamp + "]";
+ }
+}
diff --git a/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/ApexRestRequestorConsumer.java b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/ApexRestRequestorConsumer.java
new file mode 100644
index 000000000..735053279
--- /dev/null
+++ b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/ApexRestRequestorConsumer.java
@@ -0,0 +1,416 @@
+/*-
+ * ============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.restrequestor;
+
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.EnumMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import javax.ws.rs.client.Client;
+import javax.ws.rs.client.ClientBuilder;
+import javax.ws.rs.client.Entity;
+import javax.ws.rs.core.Response;
+
+import org.onap.policy.apex.core.infrastructure.threading.ApplicationThreadFactory;
+import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities;
+import org.onap.policy.apex.plugins.event.carrier.restrequestor.RESTRequestorCarrierTechnologyParameters.HTTP_METHOD;
+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.ApexEventRuntimeException;
+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;
+
+/**
+ * This class implements an Apex event consumer that issues a REST request and returns the REST response to APEX as an
+ * event.
+ *
+ * @author Liam Fallon (liam.fallon@ericsson.com)
+ */
+public class ApexRestRequestorConsumer implements ApexEventConsumer, Runnable {
+ // Get a reference to the logger
+ private static final Logger LOGGER = LoggerFactory.getLogger(ApexRestRequestorConsumer.class);
+
+ // The amount of time to wait in milliseconds between checks that the consumer thread has stopped
+ private static final long REST_REQUESTOR_WAIT_SLEEP_TIME = 50;
+
+ // The REST parameters read from the parameter service
+ private RESTRequestorCarrierTechnologyParameters restConsumerProperties;
+
+ // The timeout for REST requests
+ private long restRequestTimeout = RESTRequestorCarrierTechnologyParameters.DEFAULT_REST_REQUEST_TIMEOUT;
+
+ // The event receiver that will receive events from this consumer
+ private ApexEventReceiver eventReceiver;
+
+ // The HTTP client that makes a REST call to get an input event for Apex
+ private Client client;
+
+ // 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;
+
+ // Temporary request holder for incoming REST requests
+ private final BlockingQueue<ApexRestRequest> incomingRestRequestQueue = new LinkedBlockingQueue<>();
+
+ // Map of ongoing REST request threads indexed by the time they started at
+ private final Map<ApexRestRequest, RestRequestRunner> ongoingRestRequestMap = new ConcurrentHashMap<>();
+
+ // The number of events received to date
+ private Object eventsReceivedLock = new Object();
+ private Integer eventsReceived = 0;
+
+ // The number of the next request runner thread
+ private static long nextRequestRunnerThreadNo = 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 REST Properties
+ if (!(consumerParameters
+ .getCarrierTechnologyParameters() instanceof RESTRequestorCarrierTechnologyParameters)) {
+ final String errorMessage =
+ "specified consumer properties are not applicable to REST Requestor consumer (" + this.name + ")";
+ LOGGER.warn(errorMessage);
+ throw new ApexEventException(errorMessage);
+ }
+ restConsumerProperties =
+ (RESTRequestorCarrierTechnologyParameters) consumerParameters.getCarrierTechnologyParameters();
+
+ // Check if we are in peered mode
+ if (!consumerParameters.isPeeredMode(EventHandlerPeeredMode.REQUESTOR)) {
+ final String errorMessage = "REST Requestor consumer (" + this.name
+ + ") must run in peered requestor mode with a REST Requestor producer";
+ LOGGER.warn(errorMessage);
+ throw new ApexEventException(errorMessage);
+ }
+
+ // Check if the HTTP method has been set
+ if (restConsumerProperties.getHttpMethod() == null) {
+ restConsumerProperties
+ .setHttpMethod(RESTRequestorCarrierTechnologyParameters.DEFAULT_REQUESTOR_HTTP_METHOD);
+ }
+
+ if (!(restConsumerProperties.getHttpMethod() instanceof HTTP_METHOD)) {
+ final String errorMessage = "specified HTTP method of \"" + restConsumerProperties.getHttpMethod()
+ + "\" is invalid, only HTTP methods " + HTTP_METHOD.values()
+ + " are valid on REST Requestor consumer (" + this.name + ")";
+ LOGGER.warn(errorMessage);
+ throw new ApexEventException(errorMessage);
+ }
+
+ // Check if the HTTP URL has been set
+ if (restConsumerProperties.getURL() == null) {
+ final String errorMessage = "no URL has been specified on REST Requestor consumer (" + this.name + ")";
+ LOGGER.warn(errorMessage);
+ throw new ApexEventException(errorMessage);
+ }
+
+ // Check if the HTTP URL is valid
+ try {
+ new URL(restConsumerProperties.getURL());
+ } catch (final Exception e) {
+ final String errorMessage = "invalid URL has been specified on REST Requestor consumer (" + this.name + ")";
+ LOGGER.warn(errorMessage);
+ throw new ApexEventException(errorMessage, e);
+ }
+
+ // Set the peer timeout to the default value if its not set
+ if (consumerParameters.getPeerTimeout(EventHandlerPeeredMode.REQUESTOR) != 0) {
+ restRequestTimeout = consumerParameters.getPeerTimeout(EventHandlerPeeredMode.REQUESTOR);
+ }
+
+ // Initialize the HTTP client
+ client = ClientBuilder.newClient();
+ }
+
+ /**
+ * Receive an incoming REST request from the peered REST Requestor producer and queue it
+ *
+ * @param restRequest the incoming rest request to queue
+ * @throws ApexEventRuntimeException on queueing errors
+ */
+ public void processRestRequest(final ApexRestRequest restRequest) {
+ // Push the event onto the queue for handling
+ try {
+ incomingRestRequestQueue.add(restRequest);
+ } catch (final Exception e) {
+ final String errorMessage =
+ "could not queue request \"" + restRequest + "\" on REST Requestor consumer (" + this.name + ")";
+ LOGGER.warn(errorMessage);
+ throw new ApexEventRuntimeException(errorMessage);
+ }
+ }
+
+ /*
+ * (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;
+ }
+
+ /**
+ * Get the number of events received to date
+ *
+ * @return the number of events received
+ */
+ public int getEventsReceived() {
+ return eventsReceived;
+ }
+
+ /*
+ * (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() {
+ // The endless loop that receives events using REST calls
+ while (consumerThread.isAlive() && !stopOrderedFlag) {
+ try {
+ // Take the next event from the queue
+ final ApexRestRequest restRequest =
+ incomingRestRequestQueue.poll(REST_REQUESTOR_WAIT_SLEEP_TIME, TimeUnit.MILLISECONDS);
+ if (restRequest == null) {
+ // Poll timed out, check for request timeouts
+ timeoutExpiredRequests();
+ continue;
+ }
+
+ // Set the time stamp of the REST request
+ restRequest.setTimestamp(System.currentTimeMillis());
+
+ // Create a thread to process the REST request and place it on the map of ongoing requests
+ final RestRequestRunner restRequestRunner = new RestRequestRunner(restRequest);
+ ongoingRestRequestMap.put(restRequest, restRequestRunner);
+
+ // Start execution of the request
+ final Thread restRequestRunnerThread = new Thread(restRequestRunner);
+ restRequestRunnerThread.setName("RestRequestRunner_" + nextRequestRunnerThreadNo);
+ restRequestRunnerThread.start();
+ } catch (final InterruptedException e) {
+ LOGGER.debug("Thread interrupted, Reason {}", e.getMessage());
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ client.close();
+ }
+
+ /**
+ * This method times out REST requests that have expired
+ */
+ private void timeoutExpiredRequests() {
+ // Hold a list of timed out requests
+ final List<ApexRestRequest> timedoutRequestList = new ArrayList<>();
+
+ // Check for timeouts
+ for (final Entry<ApexRestRequest, RestRequestRunner> requestEntry : ongoingRestRequestMap.entrySet()) {
+ if (System.currentTimeMillis() - requestEntry.getKey().getTimestamp() > restRequestTimeout) {
+ requestEntry.getValue().stop();
+ timedoutRequestList.add(requestEntry.getKey());
+ }
+ }
+
+ // Interrupt timed out requests and remove them from the ongoing map
+ for (final ApexRestRequest timedoutRequest : timedoutRequestList) {
+ final String errorMessage =
+ "REST Requestor consumer (" + this.name + "), REST request timed out: " + timedoutRequest;
+ LOGGER.warn(errorMessage);
+
+ ongoingRestRequestMap.remove(timedoutRequest);
+ }
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.onap.policy.apex.apps.uservice.producer.ApexEventConsumer#stop()
+ */
+ @Override
+ public void stop() {
+ stopOrderedFlag = true;
+
+ while (consumerThread.isAlive()) {
+ ThreadUtilities.sleep(REST_REQUESTOR_WAIT_SLEEP_TIME);
+ }
+ }
+
+ /**
+ * This class is used to start a thread for each request issued.
+ *
+ * @author Liam Fallon (liam.fallon@ericsson.com)
+ */
+ private class RestRequestRunner implements Runnable {
+ private static final String APPLICATION_JSON = "application/json";
+
+ // The REST request being processed by this thread
+ private final ApexRestRequest request;
+
+ // The thread executing the REST request
+ private Thread restRequestThread;
+
+ /**
+ * Constructor, initialise the request runner with the request
+ *
+ * @param request the request this runner will issue
+ */
+ private RestRequestRunner(final ApexRestRequest request) {
+ this.request = request;
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see java.lang.Runnable#run()
+ */
+ @Override
+ public void run() {
+ // Get the thread for the request
+ restRequestThread = Thread.currentThread();
+
+ try {
+ // Execute the REST request
+ final Response response = sendEventAsRESTRequest();
+
+ // Check that the event request worked
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
+ final String errorMessage = "reception of response to \"" + request + "\" from URL \""
+ + restConsumerProperties.getURL() + "\" failed with status code " + response.getStatus()
+ + " and message \"" + response.readEntity(String.class) + "\"";
+ throw new ApexEventRuntimeException(errorMessage);
+ }
+
+ // Get the event we received
+ final String eventJSONString = response.readEntity(String.class);
+
+ // Check there is content
+ if (eventJSONString == null || eventJSONString.trim().length() == 0) {
+ final String errorMessage = "received an enpty response to \"" + request + "\" from URL \""
+ + restConsumerProperties.getURL() + "\"";
+ throw new ApexEventRuntimeException(errorMessage);
+ }
+
+ // Send the event into Apex
+ eventReceiver.receiveEvent(request.getExecutionId(), eventJSONString);
+
+ synchronized (eventsReceivedLock) {
+ eventsReceived++;
+ }
+ } catch (final Exception e) {
+ LOGGER.warn("error receiving events on thread {}", consumerThread.getName(), e);
+ } finally {
+ // Remove the request from the map of ongoing requests
+ ongoingRestRequestMap.remove(request);
+ }
+ }
+
+ /**
+ * Stop the REST request
+ */
+ private void stop() {
+ restRequestThread.interrupt();
+ }
+
+ /**
+ * Execute the REST request.
+ *
+ * @return the response to the REST request
+ */
+ public Response sendEventAsRESTRequest() {
+ switch (restConsumerProperties.getHttpMethod()) {
+ case GET:
+ return client.target(restConsumerProperties.getURL()).request(APPLICATION_JSON).get();
+
+ case PUT:
+ return client.target(restConsumerProperties.getURL()).request(APPLICATION_JSON)
+ .put(Entity.json(request.getEvent()));
+
+ case POST:
+ return client.target(restConsumerProperties.getURL()).request(APPLICATION_JSON)
+ .post(Entity.json(request.getEvent()));
+
+ case DELETE:
+ return client.target(restConsumerProperties.getURL()).request(APPLICATION_JSON).delete();
+ }
+
+ return null;
+ }
+ }
+}
diff --git a/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/ApexRestRequestorProducer.java b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/ApexRestRequestorProducer.java
new file mode 100644
index 000000000..12f194510
--- /dev/null
+++ b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/ApexRestRequestorProducer.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.restrequestor;
+
+import java.util.EnumMap;
+import java.util.Map;
+
+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.ApexEventProducer;
+import org.onap.policy.apex.service.engine.event.ApexEventRuntimeException;
+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 requestor that manages the producer side of a REST request.
+ *
+ * @author Liam Fallon (liam.fallon@ericsson.com)
+ *
+ */
+public class ApexRestRequestorProducer implements ApexEventProducer {
+ private static final Logger LOGGER = LoggerFactory.getLogger(ApexRestRequestorProducer.class);
+
+ // The REST carrier properties
+ private RESTRequestorCarrierTechnologyParameters restProducerProperties;
+
+ // 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);
+
+ // The number of events sent
+ private int eventsSent = 0;
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.onap.policy.apex.service.engine.event.ApexEventProducer#init(java.lang.String,
+ * org.onap.policy.apex.service.parameters.eventhandler.EventHandlerParameters)
+ */
+ @Override
+ public void init(final String producerName, final EventHandlerParameters producerParameters)
+ throws ApexEventException {
+ this.name = producerName;
+
+ // Check and get the REST Properties
+ if (!(producerParameters
+ .getCarrierTechnologyParameters() instanceof RESTRequestorCarrierTechnologyParameters)) {
+ final String errorMessage =
+ "specified consumer properties are not applicable to REST requestor producer (" + this.name + ")";
+ LOGGER.warn(errorMessage);
+ throw new ApexEventException(errorMessage);
+ }
+ restProducerProperties =
+ (RESTRequestorCarrierTechnologyParameters) producerParameters.getCarrierTechnologyParameters();
+
+ // Check if we are in peered mode
+ if (!producerParameters.isPeeredMode(EventHandlerPeeredMode.REQUESTOR)) {
+ final String errorMessage = "REST Requestor producer (" + this.name
+ + ") must run in peered requestor mode with a REST Requestor consumer";
+ LOGGER.warn(errorMessage);
+ throw new ApexEventException(errorMessage);
+ }
+
+ // Check if the HTTP URL has been set
+ if (restProducerProperties.getURL() != null) {
+ final String errorMessage = "URL may not be specified on REST Requestor producer (" + this.name + ")";
+ LOGGER.warn(errorMessage);
+ throw new ApexEventException(errorMessage);
+ }
+
+ // Check if the HTTP method has been set
+ if (restProducerProperties.getHttpMethod() != null) {
+ final String errorMessage =
+ "HTTP method may not be specified on REST Requestor producer (" + this.name + ")";
+ LOGGER.warn(errorMessage);
+ throw new ApexEventException(errorMessage);
+ }
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.onap.policy.apex.service.engine.event.ApexEventProducer#getName()
+ */
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ /**
+ * Get the number of events sent to date
+ *
+ * @return the number of events received
+ */
+ public int getEventsSent() {
+ return eventsSent;
+ }
+
+ /*
+ * (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.service.engine.event.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);
+ }
+
+ // Find the peered consumer for this producer
+ final PeeredReference peeredRequestorReference = peerReferenceMap.get(EventHandlerPeeredMode.REQUESTOR);
+ if (peeredRequestorReference != null) {
+ // Find the REST Response Consumer that will handle this request
+ final ApexEventConsumer consumer = peeredRequestorReference.getPeeredConsumer();
+ if (!(consumer instanceof ApexRestRequestorConsumer)) {
+ final String errorMessage = "send of event to URL \"" + restProducerProperties.getURL() + "\" failed,"
+ + " REST response consumer is not an instance of ApexRestRequestorConsumer\n" + event;
+ LOGGER.warn(errorMessage);
+ throw new ApexEventRuntimeException(errorMessage);
+ }
+
+ // Use the consumer to handle this event
+ final ApexRestRequestorConsumer restRequstConsumer = (ApexRestRequestorConsumer) consumer;
+ restRequstConsumer.processRestRequest(new ApexRestRequest(executionId, eventName, event));
+
+ eventsSent++;
+ } else {
+ // No peered consumer defined
+ final String errorMessage = "send of event to URL \"" + restProducerProperties.getURL() + "\" failed,"
+ + " REST response consumer is not defined\n" + event;
+ LOGGER.warn(errorMessage);
+ throw new ApexEventRuntimeException(errorMessage);
+ }
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.onap.policy.apex.service.engine.event.ApexEventProducer#stop()
+ */
+ @Override
+ public void stop() {
+ // For REST requestor, all the implementation is in the consumer
+ }
+}
diff --git a/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/RESTRequestorCarrierTechnologyParameters.java b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/RESTRequestorCarrierTechnologyParameters.java
new file mode 100644
index 000000000..eb5870058
--- /dev/null
+++ b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/RESTRequestorCarrierTechnologyParameters.java
@@ -0,0 +1,135 @@
+/*-
+ * ============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.restrequestor;
+
+import org.onap.policy.apex.service.parameters.carriertechnology.CarrierTechnologyParameters;
+
+/**
+ * Apex parameters for REST as an event carrier technology with Apex issuing a REST request and receiving a REST
+ * response.
+ *
+ * The parameters for this plugin are:
+ * <ol>
+ * <li>url: The URL that the Apex Rest Requestor will connect to over REST for REST request sending. This parameter is
+ * mandatory.
+ * <li>httpMethod: The HTTP method to use when making requests over REST, legal values are GET (default), POST, PUT, and
+ * DELETE.
+ * <li>restRequestTimeout: The time in milliseconds to wait for a REST request to complete.
+ * </ol>
+ *
+ * @author Liam Fallon (liam.fallon@ericsson.com)
+ */
+public class RESTRequestorCarrierTechnologyParameters extends CarrierTechnologyParameters {
+ /** The supported HTTP methods. */
+ public enum HTTP_METHOD {
+ GET, PUT, POST, DELETE
+ }
+
+ /** The label of this carrier technology. */
+ public static final String RESTREQUESTOR_CARRIER_TECHNOLOGY_LABEL = "RESTREQUESTOR";
+
+ /** The producer plugin class for the REST carrier technology. */
+ public static final String RESTREQUSTOR_EVENT_PRODUCER_PLUGIN_CLASS =
+ ApexRestRequestorProducer.class.getCanonicalName();
+
+ /** The consumer plugin class for the REST carrier technology. */
+ public static final String RESTREQUSTOR_EVENT_CONSUMER_PLUGIN_CLASS =
+ ApexRestRequestorConsumer.class.getCanonicalName();
+
+ /** The default HTTP method for request events. */
+ public static final HTTP_METHOD DEFAULT_REQUESTOR_HTTP_METHOD = HTTP_METHOD.GET;
+
+ /** The default timeout for REST requests. */
+ public static final long DEFAULT_REST_REQUEST_TIMEOUT = 500;
+
+ private String url = null;
+ private HTTP_METHOD httpMethod = null;
+
+ /**
+ * Constructor to create a REST carrier technology parameters instance and register the instance with the parameter
+ * service.
+ */
+ public RESTRequestorCarrierTechnologyParameters() {
+ super(RESTRequestorCarrierTechnologyParameters.class.getCanonicalName());
+
+ // Set the carrier technology properties for the web socket carrier technology
+ this.setLabel(RESTREQUESTOR_CARRIER_TECHNOLOGY_LABEL);
+ this.setEventProducerPluginClass(RESTREQUSTOR_EVENT_PRODUCER_PLUGIN_CLASS);
+ this.setEventConsumerPluginClass(RESTREQUSTOR_EVENT_CONSUMER_PLUGIN_CLASS);
+ }
+
+ /**
+ * Gets the URL for the REST request.
+ *
+ * @return the URL
+ */
+ public String getURL() {
+ return url;
+ }
+
+ /**
+ * Sets the URL for the REST request.
+ *
+ * @param incomingURL the URL
+ */
+ public void setURL(final String incomingURL) {
+ this.url = incomingURL;
+ }
+
+ /**
+ * Gets the HTTP method to use for the REST request.
+ *
+ * @return the HTTP method
+ */
+ public HTTP_METHOD getHttpMethod() {
+ return httpMethod;
+ }
+
+ /**
+ * Sets the HTTP method to use for the REST request.
+ *
+ * @param httpMethod the HTTP method
+ */
+ public void setHttpMethod(final HTTP_METHOD httpMethod) {
+ this.httpMethod = httpMethod;
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see java.lang.Object#toString()
+ */
+
+ @Override
+ public String toString() {
+ return "RESTRequestorCarrierTechnologyParameters [url=" + url + ", httpMethod=" + httpMethod + "]";
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.onap.policy.apex.apps.uservice.parameters.ApexParameterValidator#validate()
+ */
+ @Override
+ public String validate() {
+ return "";
+ }
+}
diff --git a/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/package-info.java b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/package-info.java
new file mode 100644
index 000000000..904de13c9
--- /dev/null
+++ b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/package-info.java
@@ -0,0 +1,28 @@
+/*-
+ * ============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=========================================================
+ */
+
+/**
+ * APEX REST consumer and producer plugins when APEX sends a REST request and receives a REST response to a REST server.
+ *
+ * @author Joss Armstrong (joss.armstrong@ericsson.com)
+ *
+ */
+
+package org.onap.policy.apex.plugins.event.carrier.restrequestor;