gerrit.onap Code Review - policy/apex-pdp.git/blob

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2016-2018 Ericsson. All rights reserved.
   4  *  Modifications Copyright (C) 2019 Nordix Foundation.
   5  * ================================================================================
   6  * Licensed under the Apache License, Version 2.0 (the "License");
   7  * you may not use this file except in compliance with the License.
   8  * You may obtain a copy of the License at
   9  *
  10  *      http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  *
  18  * SPDX-License-Identifier: Apache-2.0
  19  * ============LICENSE_END=========================================================
  20  */
  21
  22 package org.onap.policy.apex.plugins.event.carrier.restclient;
  23
  24 import java.util.EnumMap;
  25 import java.util.Map;
  26 import java.util.Properties;
  27 import java.util.Set;
  28 import java.util.Optional;
  29 import java.util.concurrent.atomic.AtomicReference;
  30
  31 import javax.ws.rs.client.Client;
  32 import javax.ws.rs.client.ClientBuilder;
  33 import javax.ws.rs.client.Entity;
  34 import javax.ws.rs.core.Response;
  35
  36 import org.onap.policy.apex.service.engine.event.ApexEventException;
  37 import org.onap.policy.apex.service.engine.event.ApexEventProducer;
  38 import org.onap.policy.apex.service.engine.event.ApexEventRuntimeException;
  39 import org.onap.policy.apex.service.engine.event.PeeredReference;
  40 import org.onap.policy.apex.service.engine.event.SynchronousEventCache;
  41 import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerParameters;
  42 import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerPeeredMode;
  43 import org.onap.policy.common.parameters.GroupValidationResult;
  44 import org.onap.policy.common.parameters.ParameterException;
  45 import org.slf4j.Logger;
  46 import org.slf4j.LoggerFactory;
  47
  48 /**
  49  * Concrete implementation of an Apex event producer that sends events using REST.
  50  *
  51  * @author Joss Armstrong (joss.armstrong@ericsson.com)
  52  *
  53  */
  54 public class ApexRestClientProducer implements ApexEventProducer {
  55     private static final Logger LOGGER = LoggerFactory.getLogger(ApexRestClientProducer.class);
  56
  57     // The HTTP client that makes a REST call with an event from Apex
  58     private Client client;
  59
  60     // The REST carrier properties
  61     private RestClientCarrierTechnologyParameters restProducerProperties;
  62
  63     // The name for this producer
  64     private String name = null;
  65
  66     // The peer references for this event handler
  67     private Map<EventHandlerPeeredMode, PeeredReference> peerReferenceMap = new EnumMap<>(EventHandlerPeeredMode.class);
  68
  69     /**
  70      * {@inheritDoc}.
  71      */
  72     @Override
  73     public void init(final String producerName, final EventHandlerParameters producerParameters)
  74             throws ApexEventException {
  75         this.name = producerName;
  76
  77         // Check and get the REST Properties
  78         if (!(producerParameters.getCarrierTechnologyParameters() instanceof RestClientCarrierTechnologyParameters)) {
  79             final String errorMessage =
  80                     "specified producer properties are not applicable to REST client producer (" + this.name + ")";
  81             LOGGER.warn(errorMessage);
  82             throw new ApexEventException(errorMessage);
  83         }
  84         restProducerProperties =
  85                 (RestClientCarrierTechnologyParameters) producerParameters.getCarrierTechnologyParameters();
  86
  87         // Check if the HTTP method has been set
  88         if (restProducerProperties.getHttpMethod() == null) {
  89             restProducerProperties.setHttpMethod(RestClientCarrierTechnologyParameters.HttpMethod.POST);
  90         }
  91
  92         if (!RestClientCarrierTechnologyParameters.HttpMethod.POST.equals(restProducerProperties.getHttpMethod())
  93                 && !RestClientCarrierTechnologyParameters.HttpMethod.PUT
  94                         .equals(restProducerProperties.getHttpMethod())) {
  95             final String errorMessage = "specified HTTP method of \"" + restProducerProperties.getHttpMethod()
  96                     + "\" is invalid, only HTTP methods \"POST\" and \"PUT\" are supproted "
  97                     + "for event sending on REST client producer (" + this.name + ")";
  98             LOGGER.warn(errorMessage);
  99             throw new ApexEventException(errorMessage);
 100         }
 101
 102         // Initialize the HTTP client
 103         client = ClientBuilder.newClient();
 104     }
 105
 106     /**
 107      * {@inheritDoc}.
 108      */
 109     @Override
 110     public String getName() {
 111         return name;
 112     }
 113
 114     /**
 115      * {@inheritDoc}.
 116      */
 117     @Override
 118     public PeeredReference getPeeredReference(final EventHandlerPeeredMode peeredMode) {
 119         return peerReferenceMap.get(peeredMode);
 120     }
 121
 122     /**
 123      * {@inheritDoc}.
 124      */
 125     @Override
 126     public void setPeeredReference(final EventHandlerPeeredMode peeredMode, final PeeredReference peeredReference) {
 127         peerReferenceMap.put(peeredMode, peeredReference);
 128     }
 129
 130     /**
 131      * {@inheritDoc}.
 132      */
 133     @Override
 134     public void sendEvent(final long executionId, final Properties executionProperties, final String eventName,
 135             final Object event) {
 136         // Check if this is a synchronized event, if so we have received a reply
 137         final SynchronousEventCache synchronousEventCache =
 138                 (SynchronousEventCache) peerReferenceMap.get(EventHandlerPeeredMode.SYNCHRONOUS);
 139         if (synchronousEventCache != null) {
 140             synchronousEventCache.removeCachedEventToApexIfExists(executionId);
 141         }
 142
 143         String untaggedUrl = restProducerProperties.getUrl();
 144         if (executionProperties != null) {
 145             Set<String> names = restProducerProperties.getKeysFromUrl();
 146             Set<String> inputProperty = executionProperties.stringPropertyNames();
 147
 148             names.stream().map(key -> Optional.of(key)).forEach(op -> {
 149                 op.filter(str -> inputProperty.contains(str))
 150                     .orElseThrow(() -> new ApexEventRuntimeException(
 151                         "key\"" + op.get() + "\"specified on url \"" + restProducerProperties.getUrl() +
 152                         "\"not found in execution properties passed by the current policy"));
 153             });
 154
 155             untaggedUrl = names.stream().reduce(untaggedUrl,
 156                 (acc, str) -> acc.replace("{" + str + "}", (String) executionProperties.get(str)));
 157         }
 158
 159         // Send the event as a REST request
 160         final Response response = sendEventAsRestRequest(untaggedUrl, (String) event);
 161
 162         // Check that the request worked
 163         if (response.getStatus() != Response.Status.OK.getStatusCode()) {
 164             final String errorMessage = "send of event to URL \"" + untaggedUrl + "\" using HTTP \""
 165                     + restProducerProperties.getHttpMethod() + "\" failed with status code " + response.getStatus()
 166                     + " and message \"" + response.readEntity(String.class) + "\", event:\n" + event;
 167             LOGGER.warn(errorMessage);
 168             throw new ApexEventRuntimeException(errorMessage);
 169         }
 170
 171         if (LOGGER.isTraceEnabled()) {
 172             LOGGER.trace("event sent from engine using {} to URL {} with HTTP {} : {} and response {} ", this.name,
 173                 untaggedUrl, restProducerProperties.getHttpMethod(), event, response);
 174         }
 175     }
 176
 177     /**
 178      * {@inheritDoc}.
 179      */
 180     @Override
 181     public void stop() {
 182         // Close the HTTP session
 183         client.close();
 184     }
 185
 186     /**
 187      * Send the event as a JSON string as a REST request.
 188      *
 189      * @param event the event to send
 190      * @return the response to the JSON request
 191      */
 192     private Response sendEventAsRestRequest(final String untaggedUrl, final String event) {
 193         // We have already checked that it is a PUT or POST request
 194         if (RestClientCarrierTechnologyParameters.HttpMethod.POST.equals(restProducerProperties.getHttpMethod())) {
 195             return client.target(untaggedUrl).request("application/json")
 196                     .headers(restProducerProperties.getHttpHeadersAsMultivaluedMap()).post(Entity.json(event));
 197         } else {
 198             return client.target(untaggedUrl).request("application/json")
 199                     .headers(restProducerProperties.getHttpHeadersAsMultivaluedMap()).put(Entity.json(event));
 200         }
 201     }
 202
 203     /**
 204      * Hook for unit test mocking of HTTP client.
 205      *
 206      * @param client the mocked client
 207      */
 208     protected void setClient(final Client client) {
 209         this.client = client;
 210     }
 211 }