From 697d02bf4e4188e3040cd987dee97d15f397a35f Mon Sep 17 00:00:00 2001 From: ramverma Date: Fri, 8 Jun 2018 11:56:21 +0100 Subject: Adding plugins-event module to apex-pdp Adding plugins-event module to apex-pdp Fix a minor bug in TextFileUtils Change-Id: I393c5f5809d078850d6669d22759ba9fa1b4f0e6 Issue-ID: POLICY-862 Signed-off-by: ramverma --- .../carrier/restrequestor/ApexRestRequest.java | 65 ++++ .../restrequestor/ApexRestRequestorConsumer.java | 416 +++++++++++++++++++++ .../restrequestor/ApexRestRequestorProducer.java | 195 ++++++++++ .../RESTRequestorCarrierTechnologyParameters.java | 135 +++++++ .../event/carrier/restrequestor/package-info.java | 28 ++ 5 files changed, 839 insertions(+) create mode 100644 plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/ApexRestRequest.java create mode 100644 plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/ApexRestRequestorConsumer.java create mode 100644 plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/ApexRestRequestorProducer.java create mode 100644 plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/RESTRequestorCarrierTechnologyParameters.java create mode 100644 plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/package-info.java (limited to 'plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main') 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 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 incomingRestRequestQueue = new LinkedBlockingQueue<>(); + + // Map of ongoing REST request threads indexed by the time they started at + private final Map 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 timedoutRequestList = new ArrayList<>(); + + // Check for timeouts + for (final Entry 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 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: + *
    + *
  1. url: The URL that the Apex Rest Requestor will connect to over REST for REST request sending. This parameter is + * mandatory. + *
  2. httpMethod: The HTTP method to use when making requests over REST, legal values are GET (default), POST, PUT, and + * DELETE. + *
  3. restRequestTimeout: The time in milliseconds to wait for a REST request to complete. + *
+ * + * @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; -- cgit 1.2.3-korg