models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperation.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * ONAP
   4  * ================================================================================
   5  * Copyright (C) 2020-2021 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.policy.controlloop.actorserviceprovider.impl;
  22
  23 import java.util.List;
  24 import java.util.concurrent.CompletableFuture;
  25 import java.util.function.BiConsumer;
  26 import lombok.Getter;
  27 import org.onap.policy.common.endpoints.utils.NetLoggerUtil.EventType;
  28 import org.onap.policy.common.utils.coder.CoderException;
  29 import org.onap.policy.common.utils.coder.StandardCoderObject;
  30 import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
  31 import org.onap.policy.controlloop.actorserviceprovider.OperationResult;
  32 import org.onap.policy.controlloop.actorserviceprovider.parameters.BidirectionalTopicConfig;
  33 import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
  34 import org.onap.policy.controlloop.actorserviceprovider.pipeline.PipelineControllerFuture;
  35 import org.onap.policy.controlloop.actorserviceprovider.topic.BidirectionalTopicHandler;
  36 import org.onap.policy.controlloop.actorserviceprovider.topic.Forwarder;
  37 import org.slf4j.Logger;
  38 import org.slf4j.LoggerFactory;
  39
  40 /**
  41  * Operation that uses a bidirectional topic.
  42  *
  43  * @param <S> response type
  44  */
  45 @Getter
  46 public abstract class BidirectionalTopicOperation<Q, S> extends OperationPartial {
  47     private static final Logger logger = LoggerFactory.getLogger(BidirectionalTopicOperation.class);
  48
  49     /**
  50      * Response status.
  51      */
  52     public enum Status {
  53         SUCCESS, FAILURE, STILL_WAITING
  54     }
  55
  56     /**
  57      * Configuration for this operation.
  58      */
  59     private final BidirectionalTopicConfig config;
  60
  61     /**
  62      * Response class.
  63      */
  64     private final Class<S> responseClass;
  65
  66     // fields extracted from "config"
  67
  68     private final BidirectionalTopicHandler topicHandler;
  69     private final Forwarder forwarder;
  70
  71
  72     /**
  73      * Constructs the object.
  74      *
  75      * @param params operation parameters
  76      * @param config configuration for this operation
  77      * @param clazz response class
  78      * @param propertyNames names of properties required by this operation
  79      */
  80     protected BidirectionalTopicOperation(ControlLoopOperationParams params, BidirectionalTopicConfig config,
  81                     Class<S> clazz, List<String> propertyNames) {
  82         super(params, config, propertyNames);
  83         this.config = config;
  84         this.responseClass = clazz;
  85         this.forwarder = config.getForwarder();
  86         this.topicHandler = config.getTopicHandler();
  87     }
  88
  89     public long getTimeoutMs() {
  90         return config.getTimeoutMs();
  91     }
  92
  93     /**
  94      * If no timeout is specified, then it returns the default timeout.
  95      */
  96     @Override
  97     protected long getTimeoutMs(Integer timeoutSec) {
  98         // TODO move this method to the superclass
  99         return (timeoutSec == null || timeoutSec == 0 ? getTimeoutMs() : super.getTimeoutMs(timeoutSec));
 100     }
 101
 102     /**
 103      * Publishes the request and arranges to receive the response.
 104      */
 105     @Override
 106     protected CompletableFuture<OperationOutcome> startOperationAsync(int attempt, OperationOutcome outcome) {
 107
 108         final var request = makeRequest(attempt);
 109         final List<String> expectedKeyValues = getExpectedKeyValues(attempt, request);
 110
 111         final PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
 112         final var executor = params.getExecutor();
 113
 114         // register a listener BEFORE publishing
 115
 116         BiConsumer<String, StandardCoderObject> listener = (rawResponse, scoResponse) -> {
 117             try {
 118                 OperationOutcome latestOutcome = processResponse(outcome, rawResponse, scoResponse);
 119                 if (latestOutcome != null) {
 120                     // final response - complete the controller
 121                     controller.completeAsync(() -> latestOutcome, executor);
 122                 }
 123             } catch (RuntimeException e) {
 124                 logger.warn("{}: failed to process response for {}", getFullName(), params.getRequestId());
 125                 controller.completeExceptionally(e);
 126             }
 127         };
 128
 129         forwarder.register(expectedKeyValues, listener);
 130
 131         // ensure listener is unregistered if the controller is canceled
 132         controller.add(() -> forwarder.unregister(expectedKeyValues, listener));
 133
 134         // publish the request
 135         try {
 136             publishRequest(request);
 137         } catch (RuntimeException e) {
 138             logger.warn("{}: failed to publish request for {}", getFullName(), params.getRequestId());
 139             forwarder.unregister(expectedKeyValues, listener);
 140             throw e;
 141         }
 142
 143         return controller;
 144     }
 145
 146     /**
 147      * Makes the request.
 148      *
 149      * @param attempt operation attempt
 150      * @return a new request
 151      */
 152     protected abstract Q makeRequest(int attempt);
 153
 154     /**
 155      * Gets values, expected in the response, that should match the selector keys.
 156      *
 157      * @param attempt operation attempt
 158      * @param request request to be published
 159      * @return a list of the values to be matched by the selector keys
 160      */
 161     protected abstract List<String> getExpectedKeyValues(int attempt, Q request);
 162
 163     /**
 164      * Publishes the request. Encodes the request, if it is not already a String.
 165      *
 166      * @param request request to be published
 167      */
 168     protected void publishRequest(Q request) {
 169         String json = prettyPrint(request);
 170         logMessage(EventType.OUT, topicHandler.getSinkTopicCommInfrastructure(), topicHandler.getSinkTopic(), json);
 171
 172         if (!topicHandler.send(json)) {
 173             throw new IllegalStateException("nothing published");
 174         }
 175     }
 176
 177     /**
 178      * Processes a response.
 179      *
 180      * @param outcome outcome to be populated
 181      * @param rawResponse raw response to process
 182      * @param scoResponse response, as a {@link StandardCoderObject}
 183      * @return the outcome, or {@code null} if still waiting for completion
 184      */
 185     protected OperationOutcome processResponse(OperationOutcome outcome, String rawResponse,
 186                     StandardCoderObject scoResponse) {
 187
 188         logger.info("{}.{}: response received for {}", params.getActor(), params.getOperation(), params.getRequestId());
 189
 190         logMessage(EventType.IN, topicHandler.getSourceTopicCommInfrastructure(), topicHandler.getSourceTopic(),
 191                         rawResponse);
 192
 193         // decode the response
 194         S response;
 195         if (responseClass == String.class) {
 196             response = responseClass.cast(rawResponse);
 197
 198         } else if (responseClass == StandardCoderObject.class) {
 199             response = responseClass.cast(scoResponse);
 200
 201         } else {
 202             try {
 203                 response = getCoder().decode(rawResponse, responseClass);
 204             } catch (CoderException e) {
 205                 logger.warn("{}.{} cannot decode response for {}", params.getActor(), params.getOperation(),
 206                                 params.getRequestId());
 207                 throw new IllegalArgumentException("cannot decode response", e);
 208             }
 209         }
 210
 211         // check its status
 212         switch (detmStatus(rawResponse, response)) {
 213             case SUCCESS:
 214                 logger.info("{}.{} request succeeded for {}", params.getActor(), params.getOperation(),
 215                                 params.getRequestId());
 216                 setOutcome(outcome, OperationResult.SUCCESS, response);
 217                 postProcessResponse(outcome, rawResponse, response);
 218                 return outcome;
 219
 220             case FAILURE:
 221                 logger.info("{}.{} request failed for {}", params.getActor(), params.getOperation(),
 222                                 params.getRequestId());
 223                 return setOutcome(outcome, OperationResult.FAILURE, response);
 224
 225             case STILL_WAITING:
 226             default:
 227                 logger.info("{}.{} request incomplete for {}", params.getActor(), params.getOperation(),
 228                                 params.getRequestId());
 229                 return null;
 230         }
 231     }
 232
 233     /**
 234      * Sets an operation's outcome and default message based on the result.
 235      *
 236      * @param outcome operation to be updated
 237      * @param result result of the operation
 238      * @param response response used to populate the outcome
 239      * @return the updated operation
 240      */
 241     public OperationOutcome setOutcome(OperationOutcome outcome, OperationResult result, S response) {
 242         outcome.setResponse(response);
 243         return setOutcome(outcome, result);
 244     }
 245
 246     /**
 247      * Processes a successful response.
 248      *
 249      * @param outcome outcome to be populated
 250      * @param rawResponse raw response
 251      * @param response decoded response
 252      */
 253     protected void postProcessResponse(OperationOutcome outcome, String rawResponse, S response) {
 254         // do nothing
 255     }
 256
 257     /**
 258      * Determines the status of the response.
 259      *
 260      * @param rawResponse raw response
 261      * @param response decoded response
 262      * @return the status of the response
 263      */
 264     protected abstract Status detmStatus(String rawResponse, S response);
 265 }