summaryrefslogtreecommitdiffstats
path: root/models-interactions
diff options
context:
space:
mode:
Diffstat (limited to 'models-interactions')
-rw-r--r--models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/BandwidthOnDemandOperator.java24
-rw-r--r--models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/RerouteOperator.java22
-rw-r--r--models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/SdncActorServiceProvider.java27
-rw-r--r--models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/SdncOperator.java132
-rw-r--r--models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/BandwidthOnDemandOperatorTest.java70
-rw-r--r--models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/BasicOperator.java94
-rw-r--r--models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/RerouteOperatorTest.java70
-rw-r--r--models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/SdncActorServiceProviderTest.java25
-rw-r--r--models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/SdncOperatorTest.java326
-rw-r--r--models-interactions/model-actors/actor.sdnc/src/test/resources/bod.json32
-rw-r--r--models-interactions/model-actors/actor.sdnc/src/test/resources/reroute.json17
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/ActorService.java7
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/AsyncResponseHandler.java119
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/CallbackManager.java84
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/OperationOutcome.java116
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/Operator.java3
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/Util.java81
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/controlloop/ControlLoopEventContext.java23
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/ActorImpl.java50
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpActor.java58
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperator.java84
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperatorPartial.java640
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/parameters/ControlLoopOperationParams.java108
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/ListenerManager.java2
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/PipelineControllerFuture.java119
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/AsyncResponseHandlerTest.java172
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/CallbackManagerTest.java89
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/OperationOutcomeTest.java137
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/UtilTest.java145
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/controlloop/ControlLoopEventContextTest.java27
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/ActorImplTest.java15
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpActorTest.java81
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperatorTest.java104
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperatorPartialTest.java792
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/ControlLoopOperationParamsTest.java128
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpActorParamsTest.java2
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpParamsTest.java2
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/PipelineControllerFutureTest.java255
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/resources/logback-test.xml4
39 files changed, 3457 insertions, 829 deletions
diff --git a/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/BandwidthOnDemandOperator.java b/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/BandwidthOnDemandOperator.java
index 0e721bf8c..2927bd85b 100644
--- a/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/BandwidthOnDemandOperator.java
+++ b/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/BandwidthOnDemandOperator.java
@@ -22,7 +22,7 @@ package org.onap.policy.controlloop.actor.sdnc;
import java.util.UUID;
import org.apache.commons.lang3.StringUtils;
-import org.onap.policy.controlloop.VirtualControlLoopEvent;
+import org.onap.policy.controlloop.actorserviceprovider.controlloop.ControlLoopEventContext;
import org.onap.policy.sdnc.SdncHealRequest;
import org.onap.policy.sdnc.SdncHealRequestHeaderInfo;
import org.onap.policy.sdnc.SdncHealRequestInfo;
@@ -37,6 +37,12 @@ import org.onap.policy.sdnc.SdncRequest;
public class BandwidthOnDemandOperator extends SdncOperator {
public static final String NAME = "BandwidthOnDemand";
+ public static final String URI = "/GENERIC-RESOURCE-API:vf-module-topology-operation";
+
+ // fields in the enrichment data
+ public static final String SERVICE_ID_KEY = "service-instance.service-instance-id";
+ public static final String VNF_ID = "vnfId";
+
/**
* Constructs the object.
*
@@ -47,19 +53,19 @@ public class BandwidthOnDemandOperator extends SdncOperator {
}
@Override
- protected SdncRequest constructRequest(VirtualControlLoopEvent onset) {
- String serviceInstance = onset.getAai().get("service-instance.service-instance-id");
+ protected SdncRequest constructRequest(ControlLoopEventContext context) {
+ String serviceInstance = context.getEnrichment().get(SERVICE_ID_KEY);
if (StringUtils.isBlank(serviceInstance)) {
- throw new IllegalArgumentException("missing enrichment data, service-instance-id");
+ throw new IllegalArgumentException("missing enrichment data, " + SERVICE_ID_KEY);
}
SdncHealVfModuleParameter bandwidth = new SdncHealVfModuleParameter();
bandwidth.setName("bandwidth");
- bandwidth.setValue(onset.getAai().get("bandwidth"));
+ bandwidth.setValue(context.getEnrichment().get("bandwidth"));
SdncHealVfModuleParameter timeStamp = new SdncHealVfModuleParameter();
timeStamp.setName("bandwidth-change-time");
- timeStamp.setValue(onset.getAai().get("bandwidth-change-time"));
+ timeStamp.setValue(context.getEnrichment().get("bandwidth-change-time"));
SdncHealVfModuleParametersInfo vfParametersInfo = new SdncHealVfModuleParametersInfo();
vfParametersInfo.addParameters(bandwidth);
@@ -80,11 +86,11 @@ public class BandwidthOnDemandOperator extends SdncOperator {
SdncRequest request = new SdncRequest();
request.setNsInstanceId(serviceInstance);
- request.setRequestId(onset.getRequestId());
- request.setUrl("/GENERIC-RESOURCE-API:vf-module-topology-operation");
+ request.setRequestId(context.getRequestId());
+ request.setUrl(URI);
SdncHealVnfInfo vnfInfo = new SdncHealVnfInfo();
- vnfInfo.setVnfId(onset.getAai().get("vnfId"));
+ vnfInfo.setVnfId(context.getEnrichment().get(VNF_ID));
SdncHealVfModuleInfo vfModuleInfo = new SdncHealVfModuleInfo();
vfModuleInfo.setVfModuleId("");
diff --git a/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/RerouteOperator.java b/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/RerouteOperator.java
index 59af31f4f..da400f8eb 100644
--- a/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/RerouteOperator.java
+++ b/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/RerouteOperator.java
@@ -22,7 +22,7 @@ package org.onap.policy.controlloop.actor.sdnc;
import java.util.UUID;
import org.apache.commons.lang3.StringUtils;
-import org.onap.policy.controlloop.VirtualControlLoopEvent;
+import org.onap.policy.controlloop.actorserviceprovider.controlloop.ControlLoopEventContext;
import org.onap.policy.sdnc.SdncHealNetworkInfo;
import org.onap.policy.sdnc.SdncHealRequest;
import org.onap.policy.sdnc.SdncHealRequestHeaderInfo;
@@ -33,6 +33,12 @@ import org.onap.policy.sdnc.SdncRequest;
public class RerouteOperator extends SdncOperator {
public static final String NAME = "Reroute";
+ public static final String URI = "/GENERIC-RESOURCE-API:network-topology-operation";
+
+ // fields in the enrichment data
+ public static final String SERVICE_ID_KEY = "service-instance.service-instance-id";
+ public static final String NETWORK_ID_KEY = "network-information.network-id";
+
/**
* Constructs the object.
*
@@ -43,17 +49,17 @@ public class RerouteOperator extends SdncOperator {
}
@Override
- protected SdncRequest constructRequest(VirtualControlLoopEvent onset) {
- String serviceInstance = onset.getAai().get("service-instance.service-instance-id");
+ protected SdncRequest constructRequest(ControlLoopEventContext context) {
+ String serviceInstance = context.getEnrichment().get(SERVICE_ID_KEY);
if (StringUtils.isBlank(serviceInstance)) {
- throw new IllegalArgumentException("missing enrichment data, service-instance-id");
+ throw new IllegalArgumentException("missing enrichment data, " + SERVICE_ID_KEY);
}
SdncHealServiceInfo serviceInfo = new SdncHealServiceInfo();
serviceInfo.setServiceInstanceId(serviceInstance);
- String networkId = onset.getAai().get("network-information.network-id");
+ String networkId = context.getEnrichment().get(NETWORK_ID_KEY);
if (StringUtils.isBlank(networkId)) {
- throw new IllegalArgumentException("missing enrichment data, network-id");
+ throw new IllegalArgumentException("missing enrichment data, " + NETWORK_ID_KEY);
}
SdncHealNetworkInfo networkInfo = new SdncHealNetworkInfo();
networkInfo.setNetworkId(networkId);
@@ -67,8 +73,8 @@ public class RerouteOperator extends SdncOperator {
SdncRequest request = new SdncRequest();
request.setNsInstanceId(serviceInstance);
- request.setRequestId(onset.getRequestId());
- request.setUrl("/GENERIC-RESOURCE-API:network-topology-operation");
+ request.setRequestId(context.getRequestId());
+ request.setUrl(URI);
SdncHealRequest healRequest = new SdncHealRequest();
healRequest.setRequestHeaderInfo(headerInfo);
diff --git a/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/SdncActorServiceProvider.java b/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/SdncActorServiceProvider.java
index 13276f929..8dc8ba50d 100644
--- a/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/SdncActorServiceProvider.java
+++ b/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/SdncActorServiceProvider.java
@@ -26,14 +26,10 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import java.util.Collections;
import java.util.List;
-import java.util.Map;
import java.util.UUID;
-import java.util.function.Function;
import org.onap.policy.controlloop.ControlLoopOperation;
import org.onap.policy.controlloop.VirtualControlLoopEvent;
-import org.onap.policy.controlloop.actorserviceprovider.Util;
-import org.onap.policy.controlloop.actorserviceprovider.impl.ActorImpl;
-import org.onap.policy.controlloop.actorserviceprovider.parameters.HttpActorParams;
+import org.onap.policy.controlloop.actorserviceprovider.impl.HttpActor;
import org.onap.policy.controlloop.policy.Policy;
import org.onap.policy.sdnc.SdncHealNetworkInfo;
import org.onap.policy.sdnc.SdncHealRequest;
@@ -49,7 +45,7 @@ import org.onap.policy.sdnc.SdncRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class SdncActorServiceProvider extends ActorImpl {
+public class SdncActorServiceProvider extends HttpActor {
private static final Logger logger = LoggerFactory.getLogger(SdncActorServiceProvider.class);
public static final String NAME = "SDNC";
@@ -78,23 +74,10 @@ public class SdncActorServiceProvider extends ActorImpl {
* Constructs the object.
*/
public SdncActorServiceProvider() {
- // @formatter:off
- super(NAME,
- new RerouteOperator(NAME),
- new BandwidthOnDemandOperator(NAME));
+ super(NAME);
- // @formatter:on
- }
-
- @Override
- protected Function<String, Map<String, Object>> makeOperatorParameters(Map<String, Object> actorParameters) {
- String actorName = getName();
-
- // @formatter:off
- return Util.translate(actorName, actorParameters, HttpActorParams.class)
- .doValidation(actorName)
- .makeOperationParameters(actorName);
- // @formatter:on
+ addOperator(new RerouteOperator(NAME));
+ addOperator(new BandwidthOnDemandOperator(NAME));
}
diff --git a/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/SdncOperator.java b/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/SdncOperator.java
index c2e4c8f8b..479ee908d 100644
--- a/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/SdncOperator.java
+++ b/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/SdncOperator.java
@@ -22,45 +22,31 @@ package org.onap.policy.controlloop.actor.sdnc;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.CompletableFuture;
import javax.ws.rs.client.Entity;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
-import lombok.AccessLevel;
-import lombok.Getter;
-import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
import org.onap.policy.common.endpoints.http.client.HttpClient;
-import org.onap.policy.common.endpoints.http.client.HttpClientFactoryInstance;
-import org.onap.policy.common.endpoints.utils.NetLoggerUtil;
-import org.onap.policy.common.endpoints.utils.NetLoggerUtil.EventType;
-import org.onap.policy.common.parameters.ValidationResult;
-import org.onap.policy.controlloop.ControlLoopOperation;
-import org.onap.policy.controlloop.VirtualControlLoopEvent;
+import org.onap.policy.common.utils.coder.CoderException;
+import org.onap.policy.common.utils.coder.StandardCoder;
+import org.onap.policy.controlloop.actorserviceprovider.AsyncResponseHandler;
+import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
import org.onap.policy.controlloop.actorserviceprovider.Util;
-import org.onap.policy.controlloop.actorserviceprovider.impl.OperatorPartial;
+import org.onap.policy.controlloop.actorserviceprovider.controlloop.ControlLoopEventContext;
+import org.onap.policy.controlloop.actorserviceprovider.impl.HttpOperator;
import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
-import org.onap.policy.controlloop.actorserviceprovider.parameters.HttpParams;
-import org.onap.policy.controlloop.actorserviceprovider.parameters.ParameterValidationRuntimeException;
import org.onap.policy.controlloop.policy.PolicyResult;
import org.onap.policy.sdnc.SdncRequest;
import org.onap.policy.sdnc.SdncResponse;
-import org.onap.policy.sdnc.util.Serialization;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Superclass for SDNC Operators.
*/
-public abstract class SdncOperator extends OperatorPartial {
+public abstract class SdncOperator extends HttpOperator {
private static final Logger logger = LoggerFactory.getLogger(SdncOperator.class);
- @Getter(AccessLevel.PROTECTED)
- private HttpClient client;
-
- /**
- * URI path for this particular operation.
- */
- private String path;
-
/**
* Constructs the object.
*
@@ -71,74 +57,92 @@ public abstract class SdncOperator extends OperatorPartial {
super(actorName, name);
}
- // TODO add a junit for this and for plug-in via ActorService
@Override
- protected void doConfigure(Map<String, Object> parameters) {
- HttpParams params = Util.translate(getFullName(), parameters, HttpParams.class);
- ValidationResult result = params.validate(getFullName());
- if (!result.isValid()) {
- throw new ParameterValidationRuntimeException("invalid parameters", result);
- }
+ protected CompletableFuture<OperationOutcome> startOperationAsync(ControlLoopOperationParams params, int attempt,
+ OperationOutcome outcome) {
- client = HttpClientFactoryInstance.getClientFactory().get(params.getClientName());
- path = params.getPath();
- }
-
- @Override
- protected ControlLoopOperation doOperation(ControlLoopOperationParams params, int attempt,
- ControlLoopOperation operation) {
-
- SdncRequest request = constructRequest(params.getContext().getEvent());
- PolicyResult result = doRequest(request);
-
- return setOutcome(params, operation, result);
+ SdncRequest request = constructRequest(params.getContext());
+ return postRequest(params, outcome, request);
}
/**
* Constructs the request.
*
- * @param onset event for which the request should be constructed
+ * @param context associated event context
* @return a new request
*/
- protected abstract SdncRequest constructRequest(VirtualControlLoopEvent onset);
+ protected abstract SdncRequest constructRequest(ControlLoopEventContext context);
/**
- * Posts the request and retrieves the response.
+ * Posts the request and and arranges to retrieve the response.
*
+ * @param params operation parameters
+ * @param outcome updated with the response
* @param sdncRequest request to be posted
* @return the result of the request
*/
- private PolicyResult doRequest(SdncRequest sdncRequest) {
+ private CompletableFuture<OperationOutcome> postRequest(ControlLoopOperationParams params, OperationOutcome outcome,
+ SdncRequest sdncRequest) {
Map<String, Object> headers = new HashMap<>();
headers.put("Accept", "application/json");
- String sdncUrl = client.getBaseUrl();
-
- String sdncRequestJson = Serialization.gsonPretty.toJson(sdncRequest);
+ String sdncUrl = getClient().getBaseUrl();
- // TODO move this into a utility
- NetLoggerUtil.log(EventType.OUT, CommInfrastructure.REST, sdncUrl, sdncRequestJson);
- logger.info("[OUT|{}|{}|]{}{}", CommInfrastructure.REST, sdncUrl, NetLoggerUtil.SYSTEM_LS, sdncRequestJson);
+ Util.logRestRequest(sdncUrl, sdncRequest);
Entity<SdncRequest> entity = Entity.entity(sdncRequest, MediaType.APPLICATION_JSON);
- // TODO modify this to use asynchronous client operations
- Response rawResponse = client.post(path, entity, headers);
- String strResponse = HttpClient.getBody(rawResponse, String.class);
+ ResponseHandler handler = new ResponseHandler(params, outcome, sdncUrl);
+ return handler.handle(getClient().post(handler, getPath(), entity, headers));
+ }
+
+ private class ResponseHandler extends AsyncResponseHandler<Response> {
+ private final String sdncUrl;
- // TODO move this into a utility
- NetLoggerUtil.log(EventType.IN, CommInfrastructure.REST, sdncUrl, strResponse);
- logger.info("[IN|{}|{}|]{}{}", "Sdnc", sdncUrl, NetLoggerUtil.SYSTEM_LS, strResponse);
- logger.info("Response to Sdnc Heal post:");
- logger.info(strResponse);
+ public ResponseHandler(ControlLoopOperationParams params, OperationOutcome outcome, String sdncUrl) {
+ super(params, outcome);
+ this.sdncUrl = sdncUrl;
+ }
- SdncResponse response = Serialization.gsonPretty.fromJson(strResponse, SdncResponse.class);
+ /**
+ * Handles the response.
+ */
+ @Override
+ protected OperationOutcome doComplete(Response rawResponse) {
+ String strResponse = HttpClient.getBody(rawResponse, String.class);
+
+ Util.logRestResponse(sdncUrl, strResponse);
+
+ SdncResponse response;
+ try {
+ response = makeDecoder().decode(strResponse, SdncResponse.class);
+ } catch (CoderException e) {
+ logger.warn("Sdnc Heal cannot decode response with http error code {}", rawResponse.getStatus(), e);
+ return SdncOperator.this.setOutcome(getParams(), getOutcome(), PolicyResult.FAILURE_EXCEPTION);
+ }
+
+ if (response.getResponseOutput() != null && "200".equals(response.getResponseOutput().getResponseCode())) {
+ return SdncOperator.this.setOutcome(getParams(), getOutcome(), PolicyResult.SUCCESS);
+
+ } else {
+ logger.info("Sdnc Heal Restcall failed with http error code {}", rawResponse.getStatus());
+ return SdncOperator.this.setOutcome(getParams(), getOutcome(), PolicyResult.FAILURE);
+ }
+ }
- if (response.getResponseOutput() == null || !"200".equals(response.getResponseOutput().getResponseCode())) {
- logger.info("Sdnc Heal Restcall failed with http error code {}", rawResponse.getStatus());
- return PolicyResult.FAILURE;
+ /**
+ * Handles exceptions.
+ */
+ @Override
+ protected OperationOutcome doFailed(Throwable thrown) {
+ logger.info("Sdnc Heal Restcall threw an exception", thrown);
+ return SdncOperator.this.setOutcome(getParams(), getOutcome(), PolicyResult.FAILURE_EXCEPTION);
}
+ }
+
+ // these may be overridden by junit tests
- return PolicyResult.SUCCESS;
+ protected StandardCoder makeDecoder() {
+ return new StandardCoder();
}
}
diff --git a/models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/BandwidthOnDemandOperatorTest.java b/models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/BandwidthOnDemandOperatorTest.java
new file mode 100644
index 000000000..02931a4f1
--- /dev/null
+++ b/models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/BandwidthOnDemandOperatorTest.java
@@ -0,0 +1,70 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2020 AT&T Intellectual Property. 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actor.sdnc;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import java.util.Map;
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.policy.common.utils.coder.CoderException;
+import org.onap.policy.sdnc.SdncRequest;
+
+public class BandwidthOnDemandOperatorTest extends BasicOperator {
+
+ private BandwidthOnDemandOperator oper;
+
+
+ /**
+ * Set up.
+ */
+ @Before
+ public void setUp() {
+ makeContext();
+ oper = new BandwidthOnDemandOperator(ACTOR);
+ }
+
+ @Test
+ public void testBandwidthOnDemandOperator() {
+ assertEquals(ACTOR, oper.getActorName());
+ assertEquals(BandwidthOnDemandOperator.NAME, oper.getName());
+ }
+
+ @Test
+ public void testConstructRequest() throws CoderException {
+ SdncRequest request = oper.constructRequest(context);
+ assertEquals("my-service", request.getNsInstanceId());
+ assertEquals(REQ_ID, request.getRequestId());
+ assertEquals(BandwidthOnDemandOperator.URI, request.getUrl());
+ assertNotNull(request.getHealRequest().getRequestHeaderInfo().getSvcRequestId());
+
+ verifyRequest("bod.json", request);
+
+ verifyMissing(oper, BandwidthOnDemandOperator.SERVICE_ID_KEY, "service");
+ }
+
+ @Override
+ protected Map<String, String> makeEnrichment() {
+ return Map.of(BandwidthOnDemandOperator.SERVICE_ID_KEY, "my-service", BandwidthOnDemandOperator.VNF_ID,
+ "my-vnf");
+ }
+}
diff --git a/models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/BasicOperator.java b/models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/BasicOperator.java
new file mode 100644
index 000000000..b9028d462
--- /dev/null
+++ b/models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/BasicOperator.java
@@ -0,0 +1,94 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2020 AT&T Intellectual Property. 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actor.sdnc;
+
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.junit.Assert.assertEquals;
+
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.UUID;
+import org.onap.policy.common.utils.coder.CoderException;
+import org.onap.policy.common.utils.coder.StandardCoder;
+import org.onap.policy.common.utils.resources.ResourceUtils;
+import org.onap.policy.controlloop.VirtualControlLoopEvent;
+import org.onap.policy.controlloop.actorserviceprovider.controlloop.ControlLoopEventContext;
+
+/**
+ * Superclass for various operator tests.
+ */
+public abstract class BasicOperator {
+ protected static final UUID REQ_ID = UUID.randomUUID();
+ protected static final String ACTOR = "my-actor";
+
+ protected Map<String, String> enrichment;
+ protected VirtualControlLoopEvent event;
+ protected ControlLoopEventContext context;
+
+ /**
+ * Pretty-prints a request and verifies that the result matches the expected JSON.
+ *
+ * @param <T> request type
+ * @param expectedJsonFile name of the file containing the expected JSON
+ * @param request request to verify
+ * @throws CoderException if the request cannot be pretty-printed
+ */
+ protected <T> void verifyRequest(String expectedJsonFile, T request) throws CoderException {
+ String json = new StandardCoder().encode(request, true);
+ String expected = ResourceUtils.getResourceAsString(expectedJsonFile);
+
+ // strip request id, because it changes each time
+ final String stripper = "svc-request-id[^,]*";
+ json = json.replaceFirst(stripper, "").trim();
+ expected = expected.replaceFirst(stripper, "").trim();
+
+ assertEquals(expected, json);
+ }
+
+ /**
+ * Verifies that an exception is thrown if a field is missing from the enrichment
+ * data.
+ *
+ * @param oper operator to construct the request
+ * @param fieldName name of the field to be removed from the enrichment data
+ * @param expectedText text expected in the exception message
+ */
+ protected void verifyMissing(SdncOperator oper, String fieldName, String expectedText) {
+ makeContext();
+ enrichment.remove(fieldName);
+
+ assertThatIllegalArgumentException().isThrownBy(() -> oper.constructRequest(context))
+ .withMessageContaining("missing").withMessageContaining(expectedText);
+ }
+
+ protected void makeContext() {
+ // need a mutable map, so make a copy
+ enrichment = new TreeMap<>(makeEnrichment());
+
+ event = new VirtualControlLoopEvent();
+ event.setRequestId(REQ_ID);
+ event.setAai(enrichment);
+
+ context = new ControlLoopEventContext(event);
+ }
+
+ protected abstract Map<String, String> makeEnrichment();
+}
diff --git a/models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/RerouteOperatorTest.java b/models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/RerouteOperatorTest.java
new file mode 100644
index 000000000..0a7bcad6f
--- /dev/null
+++ b/models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/RerouteOperatorTest.java
@@ -0,0 +1,70 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2020 AT&T Intellectual Property. 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actor.sdnc;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import java.util.Map;
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.policy.common.utils.coder.CoderException;
+import org.onap.policy.sdnc.SdncRequest;
+
+public class RerouteOperatorTest extends BasicOperator {
+
+ private RerouteOperator oper;
+
+
+ /**
+ * Set up.
+ */
+ @Before
+ public void setUp() {
+ makeContext();
+ oper = new RerouteOperator(ACTOR);
+ }
+
+ @Test
+ public void testRerouteOperator() {
+ assertEquals(ACTOR, oper.getActorName());
+ assertEquals(RerouteOperator.NAME, oper.getName());
+ }
+
+ @Test
+ public void testConstructRequest() throws CoderException {
+ SdncRequest request = oper.constructRequest(context);
+ assertEquals("my-service", request.getNsInstanceId());
+ assertEquals(REQ_ID, request.getRequestId());
+ assertEquals(RerouteOperator.URI, request.getUrl());
+ assertNotNull(request.getHealRequest().getRequestHeaderInfo().getSvcRequestId());
+
+ verifyRequest("reroute.json", request);
+
+ verifyMissing(oper, RerouteOperator.SERVICE_ID_KEY, "service");
+ verifyMissing(oper, RerouteOperator.NETWORK_ID_KEY, "network");
+ }
+
+ @Override
+ protected Map<String, String> makeEnrichment() {
+ return Map.of(RerouteOperator.SERVICE_ID_KEY, "my-service", RerouteOperator.NETWORK_ID_KEY, "my-network");
+ }
+}
diff --git a/models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/SdncActorServiceProviderTest.java b/models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/SdncActorServiceProviderTest.java
index 9739c7145..08655c349 100644
--- a/models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/SdncActorServiceProviderTest.java
+++ b/models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/SdncActorServiceProviderTest.java
@@ -3,7 +3,7 @@
* TestSdncActorServiceProvider
* ================================================================================
* Copyright (C) 2018-2019 Huawei. All rights reserved.
- * Modifications Copyright (C) 2018-2019 AT&T Corp. All rights reserved.
+ * Modifications Copyright (C) 2018-2020 AT&T Corp. All rights reserved.
* Modifications Copyright (C) 2019 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
@@ -26,8 +26,10 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
+import java.util.Arrays;
import java.util.Objects;
import java.util.UUID;
+import java.util.stream.Collectors;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -36,19 +38,19 @@ import org.onap.policy.controlloop.ControlLoopOperation;
import org.onap.policy.controlloop.VirtualControlLoopEvent;
import org.onap.policy.controlloop.policy.Policy;
import org.onap.policy.sdnc.SdncRequest;
-import org.onap.policy.simulators.Util;
public class SdncActorServiceProviderTest {
- private static final String REROUTE = "Reroute";
+ private static final String REROUTE = RerouteOperator.NAME;
/**
* Set up before test class.
+ *
* @throws Exception if the A&AI simulator cannot be started
*/
@BeforeClass
public static void setUpSimulator() throws Exception {
- Util.buildAaiSim();
+ org.onap.policy.simulators.Util.buildAaiSim();
}
@AfterClass
@@ -57,6 +59,18 @@ public class SdncActorServiceProviderTest {
}
@Test
+ public void testSdncActorServiceProvider() {
+ final SdncActorServiceProvider prov = new SdncActorServiceProvider();
+
+ // verify that it has the operators we expect
+ var expected = Arrays.asList(BandwidthOnDemandOperator.NAME, RerouteOperator.NAME).stream().sorted()
+ .collect(Collectors.toList());
+ var actual = prov.getOperationNames().stream().sorted().collect(Collectors.toList());
+
+ assertEquals(expected.toString(), actual.toString());
+ }
+
+ @Test
public void testConstructRequest() {
VirtualControlLoopEvent onset = new VirtualControlLoopEvent();
ControlLoopOperation operation = new ControlLoopOperation();
@@ -84,8 +98,7 @@ public class SdncActorServiceProviderTest {
policy.setRecipe(REROUTE);
assertNotNull(provider.constructRequest(onset, operation, policy));
- SdncRequest request =
- provider.constructRequest(onset, operation, policy);
+ SdncRequest request = provider.constructRequest(onset, operation, policy);
assertEquals(requestId, Objects.requireNonNull(request).getRequestId());
assertEquals("reoptimize", request.getHealRequest().getRequestHeaderInfo().getSvcAction());
diff --git a/models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/SdncOperatorTest.java b/models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/SdncOperatorTest.java
new file mode 100644
index 000000000..25d383eb8
--- /dev/null
+++ b/models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/SdncOperatorTest.java
@@ -0,0 +1,326 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2020 AT&T Intellectual Property. 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actor.sdnc;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.core.Response.Status;
+import lombok.Setter;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams.TopicParamsBuilder;
+import org.onap.policy.common.endpoints.http.client.HttpClient;
+import org.onap.policy.common.endpoints.http.client.HttpClientFactoryInstance;
+import org.onap.policy.common.endpoints.http.server.HttpServletServer;
+import org.onap.policy.common.endpoints.http.server.HttpServletServerFactoryInstance;
+import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
+import org.onap.policy.common.gson.GsonMessageBodyHandler;
+import org.onap.policy.common.utils.coder.CoderException;
+import org.onap.policy.common.utils.coder.StandardCoder;
+import org.onap.policy.common.utils.network.NetworkUtil;
+import org.onap.policy.controlloop.VirtualControlLoopEvent;
+import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
+import org.onap.policy.controlloop.actorserviceprovider.Util;
+import org.onap.policy.controlloop.actorserviceprovider.controlloop.ControlLoopEventContext;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.HttpParams;
+import org.onap.policy.controlloop.policy.PolicyResult;
+import org.onap.policy.sdnc.SdncHealRequest;
+import org.onap.policy.sdnc.SdncRequest;
+import org.onap.policy.sdnc.SdncResponse;
+import org.onap.policy.sdnc.SdncResponseOutput;
+
+public class SdncOperatorTest {
+ public static final String MEDIA_TYPE_APPLICATION_JSON = "application/json";
+ private static final String EXPECTED_EXCEPTION = "expected exception";
+ public static final String HTTP_CLIENT = "my-http-client";
+ public static final String HTTP_NO_SERVER = "my-http-no-server-client";
+ private static final String ACTOR = "my-actor";
+ private static final String OPERATION = "my-operation";
+
+ /**
+ * Outcome to be added to the response.
+ */
+ @Setter
+ private static SdncResponseOutput output;
+
+
+ private VirtualControlLoopEvent event;
+ private ControlLoopEventContext context;
+ private MyOper oper;
+
+ /**
+ * Starts the SDNC simulator.
+ */
+ @BeforeClass
+ public static void setUpBeforeClass() throws Exception {
+ // allocate a port
+ int port = NetworkUtil.allocPort();
+
+ /*
+ * Start the simulator. Must use "Properties" to configure it, otherwise the
+ * server will use the wrong serialization provider.
+ */
+ Properties svrprops = getServerProperties("my-server", port);
+ HttpServletServerFactoryInstance.getServerFactory().build(svrprops).forEach(HttpServletServer::start);
+
+ /*
+ * Start the clients, one to the server, and one to a non-existent server.
+ */
+ TopicParamsBuilder builder = BusTopicParams.builder().managed(true).hostname("localhost").basePath("sdnc")
+ .serializationProvider(GsonMessageBodyHandler.class.getName());
+
+ HttpClientFactoryInstance.getClientFactory().build(builder.clientName(HTTP_CLIENT).port(port).build());
+
+ HttpClientFactoryInstance.getClientFactory()
+ .build(builder.clientName(HTTP_NO_SERVER).port(NetworkUtil.allocPort()).build());
+ }
+
+ @AfterClass
+ public static void tearDownAfterClass() {
+ HttpClientFactoryInstance.getClientFactory().destroy();
+ HttpServletServerFactoryInstance.getServerFactory().destroy();
+ }
+
+ /**
+ * Initializes {@link #oper} and sets {@link #output} to a success code.
+ */
+ @Before
+ public void setUp() {
+ event = new VirtualControlLoopEvent();
+ context = new ControlLoopEventContext(event);
+
+ initOper(HTTP_CLIENT);
+
+ output = new SdncResponseOutput();
+ output.setResponseCode("200");
+ }
+
+ @After
+ public void tearDown() {
+ oper.shutdown();
+ }
+
+ @Test
+ public void testSdncOperator() {
+ assertEquals(ACTOR, oper.getActorName());
+ assertEquals(OPERATION, oper.getName());
+ assertEquals(ACTOR + "." + OPERATION, oper.getFullName());
+ }
+
+ @Test
+ public void testGetClient() {
+ assertNotNull(oper.getTheClient());
+ }
+
+ @Test
+ public void testStartOperationAsync_testPostRequest() throws Exception {
+ OperationOutcome outcome = runOperation();
+ assertNotNull(outcome);
+ assertEquals(PolicyResult.SUCCESS, outcome.getResult());
+ }
+
+ /**
+ * Tests postRequest() when decode() throws an exception.
+ */
+ @Test
+ public void testPostRequestDecodeException() throws Exception {
+
+ oper.setDecodeFailure(true);
+
+ OperationOutcome outcome = runOperation();
+ assertNotNull(outcome);
+ assertEquals(PolicyResult.FAILURE_EXCEPTION, outcome.getResult());
+ }
+
+ /**
+ * Tests postRequest() when there is no "output" field in the response.
+ */
+ @Test
+ public void testPostRequestNoOutput() throws Exception {
+
+ setOutput(null);
+
+ OperationOutcome outcome = runOperation();
+ assertNotNull(outcome);
+ assertEquals(PolicyResult.FAILURE, outcome.getResult());
+ }
+
+ /**
+ * Tests postRequest() when the output is not a success.
+ */
+ @Test
+ public void testPostRequestOutputFailure() throws Exception {
+
+ output.setResponseCode(null);
+
+ OperationOutcome outcome = runOperation();
+ assertNotNull(outcome);
+ assertEquals(PolicyResult.FAILURE, outcome.getResult());
+ }
+
+ /**
+ * Tests postRequest() when the post() request throws an exception retrieving the
+ * response.
+ */
+ @Test
+ public void testPostRequestException() throws Exception {
+
+ // reset "oper" to point to a non-existent server
+ oper.shutdown();
+ initOper(HTTP_NO_SERVER);
+
+ OperationOutcome outcome = runOperation();
+ assertNotNull(outcome);
+ assertEquals(PolicyResult.FAILURE_EXCEPTION, outcome.getResult());
+ }
+
+ private static Properties getServerProperties(String name, int port) {
+ final Properties props = new Properties();
+ props.setProperty(PolicyEndPointProperties.PROPERTY_HTTP_SERVER_SERVICES, name);
+
+ final String svcpfx = PolicyEndPointProperties.PROPERTY_HTTP_SERVER_SERVICES + "." + name;
+
+ props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_REST_CLASSES_SUFFIX, Server.class.getName());
+ props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_HOST_SUFFIX, "localhost");
+ props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_PORT_SUFFIX, String.valueOf(port));
+ props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX, "true");
+ props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_SWAGGER_SUFFIX, "false");
+
+ props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_SERIALIZATION_PROVIDER,
+ GsonMessageBodyHandler.class.getName());
+ return props;
+ }
+
+ /**
+ * Initializes {@link #oper}.
+ *
+ * @param clientName name of the client which it should use
+ */
+ private void initOper(String clientName) {
+ oper = new MyOper();
+
+ HttpParams params = HttpParams.builder().clientName(clientName).path("request").build();
+ Map<String, Object> mapParams = Util.translateToMap(OPERATION, params);
+ oper.configure(mapParams);
+ oper.start();
+ }
+
+ /**
+ * Runs the operation.
+ *
+ * @return the outcome of the operation, or {@code null} if it does not complete in
+ * time
+ */
+ private OperationOutcome runOperation() throws InterruptedException, ExecutionException, TimeoutException {
+ ControlLoopOperationParams params =
+ ControlLoopOperationParams.builder().actor(ACTOR).operation(OPERATION).context(context).build();
+
+ CompletableFuture<OperationOutcome> future = oper.startOperationAsync(params, 1, params.makeOutcome());
+
+ return future.get(5, TimeUnit.SECONDS);
+ }
+
+
+ private class MyOper extends SdncOperator {
+
+ /**
+ * Set to {@code true} to cause the decoder to throw an exception.
+ */
+ @Setter
+ private boolean decodeFailure = false;
+
+ public MyOper() {
+ super(ACTOR, OPERATION);
+ }
+
+ protected HttpClient getTheClient() {
+ return getClient();
+ }
+
+ @Override
+ protected SdncRequest constructRequest(ControlLoopEventContext context) {
+ SdncRequest request = new SdncRequest();
+
+ SdncHealRequest heal = new SdncHealRequest();
+ request.setHealRequest(heal);
+
+ return request;
+ }
+
+ @Override
+ protected StandardCoder makeDecoder() {
+ if (decodeFailure) {
+ // return a coder that throws exceptions when decode() is invoked
+ return new StandardCoder() {
+ @Override
+ public <T> T decode(String json, Class<T> clazz) throws CoderException {
+ throw new CoderException(EXPECTED_EXCEPTION);
+ }
+ };
+
+ } else {
+ return super.makeDecoder();
+ }
+ }
+ }
+
+ /**
+ * SDNC Simulator.
+ */
+ @Path("/sdnc")
+ @Produces(MEDIA_TYPE_APPLICATION_JSON)
+ public static class Server {
+
+ /**
+ * Generates a response.
+ *
+ * @param request incoming request
+ * @return resulting response
+ */
+ @POST
+ @Path("/request")
+ @Consumes(value = {MEDIA_TYPE_APPLICATION_JSON})
+ public Response postRequest(SdncRequest request) {
+
+ SdncResponse response = new SdncResponse();
+ response.setResponseOutput(output);
+
+ return Response.status(Status.OK).entity(response).build();
+ }
+ }
+}
diff --git a/models-interactions/model-actors/actor.sdnc/src/test/resources/bod.json b/models-interactions/model-actors/actor.sdnc/src/test/resources/bod.json
new file mode 100644
index 000000000..8c60bbdd8
--- /dev/null
+++ b/models-interactions/model-actors/actor.sdnc/src/test/resources/bod.json
@@ -0,0 +1,32 @@
+{
+ "input": {
+ "sdnc-request-header": {
+ "svc-request-id": "076b243e-9236-4409-973d-dd318dcab3e9",
+ "svc-action": "update"
+ },
+ "request-information": {
+ "request-action": "SdwanBandwidthChange"
+ },
+ "service-information": {
+ "service-instance-id": "my-service"
+ },
+ "vnf-information": {
+ "vnf-id": "my-vnf"
+ },
+ "vf-module-information": {
+ "vf-module-id": ""
+ },
+ "vf-module-request-input": {
+ "vf-module-input-parameters": {
+ "param": [
+ {
+ "name": "bandwidth"
+ },
+ {
+ "name": "bandwidth-change-time"
+ }
+ ]
+ }
+ }
+ }
+}
diff --git a/models-interactions/model-actors/actor.sdnc/src/test/resources/reroute.json b/models-interactions/model-actors/actor.sdnc/src/test/resources/reroute.json
new file mode 100644
index 000000000..da70a55a3
--- /dev/null
+++ b/models-interactions/model-actors/actor.sdnc/src/test/resources/reroute.json
@@ -0,0 +1,17 @@
+{
+ "input": {
+ "sdnc-request-header": {
+ "svc-request-id": "66087b39-8606-4367-9ac8-bdf64e162f29",
+ "svc-action": "reoptimize"
+ },
+ "request-information": {
+ "request-action": "ReoptimizeSOTNInstance"
+ },
+ "service-information": {
+ "service-instance-id": "my-service"
+ },
+ "network-information": {
+ "network-id": "my-network"
+ }
+ }
+}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/ActorService.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/ActorService.java
index 13f09b1ad..2886b1feb 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/ActorService.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/ActorService.java
@@ -63,7 +63,6 @@ public class ActorService extends StartConfigPartial<Map<String, Object>> {
return newActor;
}
- // TODO: should this throw an exception?
logger.warn("duplicate actor names for {}: {}, ignoring {}", name,
existingActor.getClass().getSimpleName(), newActor.getClass().getSimpleName());
return existingActor;
@@ -158,7 +157,7 @@ public class ActorService extends StartConfigPartial<Map<String, Object>> {
for (Actor actor : name2actor.values()) {
if (actor.isConfigured()) {
- Util.logException(actor::start, "failed to start actor {}", actor.getName());
+ Util.runFunction(actor::start, "failed to start actor {}", actor.getName());
} else {
logger.warn("not starting unconfigured actor {}", actor.getName());
@@ -170,7 +169,7 @@ public class ActorService extends StartConfigPartial<Map<String, Object>> {
protected void doStop() {
logger.info("stopping actors");
name2actor.values()
- .forEach(actor -> Util.logException(actor::stop, "failed to stop actor {}", actor.getName()));
+ .forEach(actor -> Util.runFunction(actor::stop, "failed to stop actor {}", actor.getName()));
}
@Override
@@ -179,7 +178,7 @@ public class ActorService extends StartConfigPartial<Map<String, Object>> {
// @formatter:off
name2actor.values().forEach(
- actor -> Util.logException(actor::shutdown, "failed to shutdown actor {}", actor.getName()));
+ actor -> Util.runFunction(actor::shutdown, "failed to shutdown actor {}", actor.getName()));
// @formatter:on
}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/AsyncResponseHandler.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/AsyncResponseHandler.java
new file mode 100644
index 000000000..d78403809
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/AsyncResponseHandler.java
@@ -0,0 +1,119 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2020 AT&T Intellectual Property. 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actorserviceprovider;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
+import javax.ws.rs.client.InvocationCallback;
+import lombok.AccessLevel;
+import lombok.Getter;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
+import org.onap.policy.controlloop.actorserviceprovider.pipeline.PipelineControllerFuture;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Handler for a <i>single</i> asynchronous response.
+ *
+ * @param <T> response type
+ */
+@Getter
+public abstract class AsyncResponseHandler<T> implements InvocationCallback<T> {
+
+ private static final Logger logger = LoggerFactory.getLogger(AsyncResponseHandler.class);
+
+ @Getter(AccessLevel.NONE)
+ private final PipelineControllerFuture<OperationOutcome> result = new PipelineControllerFuture<>();
+ private final ControlLoopOperationParams params;
+ private final OperationOutcome outcome;
+
+ /**
+ * Constructs the object.
+ *
+ * @param params operation parameters
+ * @param outcome outcome to be populated based on the response
+ */
+ public AsyncResponseHandler(ControlLoopOperationParams params, OperationOutcome outcome) {
+ this.params = params;
+ this.outcome = outcome;
+ }
+
+ /**
+ * Handles the given future, arranging to cancel it when the response is received.
+ *
+ * @param future future to be handled
+ * @return a future to be used to cancel or wait for the response
+ */
+ public CompletableFuture<OperationOutcome> handle(Future<T> future) {
+ result.add(future);
+ return result;
+ }
+
+ /**
+ * Invokes {@link #doComplete()} and then completes "this" with the returned value.
+ */
+ @Override
+ public void completed(T rawResponse) {
+ try {
+ logger.trace("{}.{}: response completed for {}", params.getActor(), params.getOperation(),
+ params.getRequestId());
+ result.complete(doComplete(rawResponse));
+
+ } catch (RuntimeException e) {
+ logger.trace("{}.{}: response handler threw an exception for {}", params.getActor(), params.getOperation(),
+ params.getRequestId());
+ result.completeExceptionally(e);
+ }
+ }
+
+ /**
+ * Invokes {@link #doFailed()} and then completes "this" with the returned value.
+ */
+ @Override
+ public void failed(Throwable throwable) {
+ try {
+ logger.trace("{}.{}: response failure for {}", params.getActor(), params.getOperation(),
+ params.getRequestId());
+ result.complete(doFailed(throwable));
+
+ } catch (RuntimeException e) {
+ logger.trace("{}.{}: response failure handler threw an exception for {}", params.getActor(),
+ params.getOperation(), params.getRequestId());
+ result.completeExceptionally(e);
+ }
+ }
+
+ /**
+ * Completes the processing of a response.
+ *
+ * @param rawResponse raw response that was received
+ * @return the outcome
+ */
+ protected abstract OperationOutcome doComplete(T rawResponse);
+
+ /**
+ * Handles a response exception.
+ *
+ * @param thrown exception that was thrown
+ * @return the outcome
+ */
+ protected abstract OperationOutcome doFailed(Throwable thrown);
+}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/CallbackManager.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/CallbackManager.java
new file mode 100644
index 000000000..7d7c1d902
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/CallbackManager.java
@@ -0,0 +1,84 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2020 AT&T Intellectual Property. 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actorserviceprovider;
+
+import java.time.Instant;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Manager for "start" and "end" callbacks.
+ */
+public class CallbackManager implements Runnable {
+ private final AtomicReference<Instant> startTime = new AtomicReference<>();
+ private final AtomicReference<Instant> endTime = new AtomicReference<>();
+
+ /**
+ * Determines if the "start" callback can be invoked. If so, it sets the
+ * {@link #startTime} to the current time.
+ *
+ * @return {@code true} if the "start" callback can be invoked, {@code false}
+ * otherwise
+ */
+ public boolean canStart() {
+ return startTime.compareAndSet(null, Instant.now());
+ }
+
+ /**
+ * Determines if the "end" callback can be invoked. If so, it sets the
+ * {@link #endTime} to the current time.
+ *
+ * @return {@code true} if the "end" callback can be invoked, {@code false}
+ * otherwise
+ */
+ public boolean canEnd() {
+ return endTime.compareAndSet(null, Instant.now());
+ }
+
+ /**
+ * Gets the start time.
+ *
+ * @return the start time, or {@code null} if {@link #canStart()} has not been
+ * invoked yet.
+ */
+ public Instant getStartTime() {
+ return startTime.get();
+ }
+
+ /**
+ * Gets the end time.
+ *
+ * @return the end time, or {@code null} if {@link #canEnd()} has not been invoked
+ * yet.
+ */
+ public Instant getEndTime() {
+ return endTime.get();
+ }
+
+ /**
+ * Prevents further callbacks from being executed by setting {@link #startTime}
+ * and {@link #endTime}.
+ */
+ @Override
+ public void run() {
+ canStart();
+ canEnd();
+ }
+}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/OperationOutcome.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/OperationOutcome.java
new file mode 100644
index 000000000..6b0924807
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/OperationOutcome.java
@@ -0,0 +1,116 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2020 AT&T Intellectual Property. 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actorserviceprovider;
+
+import java.time.Instant;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.NonNull;
+import org.onap.policy.controlloop.ControlLoopOperation;
+import org.onap.policy.controlloop.policy.PolicyResult;
+
+/**
+ * Outcome from an operation. Objects of this type are passed from one stage to the next.
+ */
+@Data
+@NoArgsConstructor
+public class OperationOutcome {
+ private String actor;
+ private String operation;
+ private String target;
+ private Instant start;
+ private Instant end;
+ private String subRequestId;
+ private PolicyResult result = PolicyResult.SUCCESS;
+ private String message;
+
+ /**
+ * Copy constructor.
+ *
+ * @param source source object from which to copy
+ */
+ public OperationOutcome(OperationOutcome source) {
+ this.actor = source.actor;
+ this.operation = source.operation;
+ this.target = source.target;
+ this.start = source.start;
+ this.end = source.end;
+ this.subRequestId = source.subRequestId;
+ this.result = source.result;
+ this.message = source.message;
+ }
+
+ /**
+ * Creates a {@link ControlLoopOperation}, populating all fields with the values from
+ * this object. Sets the outcome field to the string representation of this object's
+ * outcome.
+ *
+ * @return
+ */
+ public ControlLoopOperation toControlLoopOperation() {
+ ControlLoopOperation clo = new ControlLoopOperation();
+
+ clo.setActor(actor);
+ clo.setOperation(operation);
+ clo.setTarget(target);
+ clo.setStart(start);
+ clo.setEnd(end);
+ clo.setSubRequestId(subRequestId);
+ clo.setOutcome(result.toString());
+ clo.setMessage(message);
+
+ return clo;
+ }
+
+ /**
+ * Determines if this outcome is for the given actor and operation.
+ *
+ * @param actor actor name
+ * @param operation operation name
+ * @return {@code true} if this outcome is for the given actor and operation
+ */
+ public boolean isFor(@NonNull String actor, @NonNull String operation) {
+ // do the operation check first, as it's most likely to be unique
+ return (operation.equals(this.operation) && actor.equals(this.actor));
+ }
+
+ /**
+ * Determines if an outcome is for the given actor and operation.
+ *
+ * @param outcome outcome to be examined, or {@code null}
+ * @param actor actor name
+ * @param operation operation name
+ * @return {@code true} if this outcome is for the given actor and operation,
+ * {@code false} it is {@code null} or not for the actor/operation
+ */
+ public static boolean isFor(OperationOutcome outcome, String actor, String operation) {
+ return (outcome != null && outcome.isFor(actor, operation));
+ }
+
+ /**
+ * Sets the result.
+ *
+ * @param result new result
+ */
+ public void setResult(@NonNull PolicyResult result) {
+ this.result = result;
+ }
+}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/Operator.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/Operator.java
index e308ee42e..c09460e34 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/Operator.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/Operator.java
@@ -24,7 +24,6 @@ import java.util.Map;
import java.util.concurrent.CompletableFuture;
import org.onap.policy.common.capabilities.Configurable;
import org.onap.policy.common.capabilities.Startable;
-import org.onap.policy.controlloop.ControlLoopOperation;
import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
/**
@@ -54,5 +53,5 @@ public interface Operator extends Startable, Configurable<Map<String, Object>> {
* @param params parameters needed to start the operation
* @return a future that can be used to cancel or await the result of the operation
*/
- CompletableFuture<ControlLoopOperation> startOperation(ControlLoopOperationParams params);
+ CompletableFuture<OperationOutcome> startOperation(ControlLoopOperationParams params);
}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/Util.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/Util.java
index 0aba1a7fa..c3ddd17f3 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/Util.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/Util.java
@@ -23,6 +23,9 @@ package org.onap.policy.controlloop.actorserviceprovider;
import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.Map;
+import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
+import org.onap.policy.common.endpoints.utils.NetLoggerUtil;
+import org.onap.policy.common.endpoints.utils.NetLoggerUtil.EventType;
import org.onap.policy.common.utils.coder.Coder;
import org.onap.policy.common.utils.coder.CoderException;
import org.onap.policy.common.utils.coder.StandardCoder;
@@ -53,6 +56,82 @@ public class Util {
}
/**
+ * Logs a REST request. If the request is not of type, String, then it attempts to
+ * pretty-print it into JSON before logging.
+ *
+ * @param url request URL
+ * @param request request to be logged
+ */
+ public static <T> void logRestRequest(String url, T request) {
+ logRestRequest(new StandardCoder(), url, request);
+ }
+
+ /**
+ * Logs a REST request. If the request is not of type, String, then it attempts to
+ * pretty-print it into JSON before logging.
+ *
+ * @param coder coder to be used to pretty-print the request
+ * @param url request URL
+ * @param request request to be logged
+ */
+ protected static <T> void logRestRequest(Coder coder, String url, T request) {
+ String json;
+ try {
+ if (request instanceof String) {
+ json = request.toString();
+ } else {
+ json = coder.encode(request, true);
+ }
+
+ } catch (CoderException e) {
+ logger.warn("cannot pretty-print request", e);
+ json = request.toString();
+ }
+
+ NetLoggerUtil.log(EventType.OUT, CommInfrastructure.REST, url, json);
+ logger.info("[OUT|{}|{}|]{}{}", CommInfrastructure.REST, url, NetLoggerUtil.SYSTEM_LS, json);
+ }
+
+ /**
+ * Logs a REST response. If the response is not of type, String, then it attempts to
+ * pretty-print it into JSON before logging.
+ *
+ * @param url request URL
+ * @param response response to be logged
+ */
+ public static <T> void logRestResponse(String url, T response) {
+ logRestResponse(new StandardCoder(), url, response);
+ }
+
+ /**
+ * Logs a REST response. If the request is not of type, String, then it attempts to
+ * pretty-print it into JSON before logging.
+ *
+ * @param coder coder to be used to pretty-print the response
+ * @param url request URL
+ * @param response response to be logged
+ */
+ protected static <T> void logRestResponse(Coder coder, String url, T response) {
+ String json;
+ try {
+ if (response == null) {
+ json = null;
+ } else if (response instanceof String) {
+ json = response.toString();
+ } else {
+ json = coder.encode(response, true);
+ }
+
+ } catch (CoderException e) {
+ logger.warn("cannot pretty-print response", e);
+ json = response.toString();
+ }
+
+ NetLoggerUtil.log(EventType.IN, CommInfrastructure.REST, url, json);
+ logger.info("[IN|{}|{}|]{}{}", CommInfrastructure.REST, url, NetLoggerUtil.SYSTEM_LS, json);
+ }
+
+ /**
* Runs a function and logs a message if it throws an exception. Does <i>not</i>
* re-throw the exception.
*
@@ -60,7 +139,7 @@ public class Util {
* @param exceptionMessage message to log if an exception is thrown
* @param exceptionArgs arguments to be passed to the logger
*/
- public static void logException(Runnable function, String exceptionMessage, Object... exceptionArgs) {
+ public static void runFunction(Runnable function, String exceptionMessage, Object... exceptionArgs) {
try {
function.run();
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/controlloop/ControlLoopEventContext.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/controlloop/ControlLoopEventContext.java
index 68bbe7edc..cd4d2570f 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/controlloop/ControlLoopEventContext.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/controlloop/ControlLoopEventContext.java
@@ -22,11 +22,12 @@ package org.onap.policy.controlloop.actorserviceprovider.controlloop;
import java.io.Serializable;
import java.util.Map;
+import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import lombok.AccessLevel;
import lombok.Getter;
+import lombok.NonNull;
import lombok.Setter;
-import org.onap.policy.aai.AaiCqResponse;
import org.onap.policy.controlloop.VirtualControlLoopEvent;
/**
@@ -37,23 +38,35 @@ import org.onap.policy.controlloop.VirtualControlLoopEvent;
public class ControlLoopEventContext implements Serializable {
private static final long serialVersionUID = 1L;
- @Setter(AccessLevel.NONE)
+
private final VirtualControlLoopEvent event;
- private AaiCqResponse aaiCqResponse;
+ /**
+ * Enrichment data extracted from the event. Never {@code null}, though it may be
+ * immutable.
+ */
+ private final Map<String, String> enrichment;
- // TODO may remove this if it proves not to be needed
@Getter(AccessLevel.NONE)
@Setter(AccessLevel.NONE)
private Map<String, Serializable> properties = new ConcurrentHashMap<>();
/**
+ * Request ID extracted from the event, or a generated value if the event has no
+ * request id; never {@code null}.
+ */
+ private final UUID requestId;
+
+
+ /**
* Constructs the object.
*
* @param event event with which this is associated
*/
- public ControlLoopEventContext(VirtualControlLoopEvent event) {
+ public ControlLoopEventContext(@NonNull VirtualControlLoopEvent event) {
this.event = event;
+ this.requestId = (event.getRequestId() != null ? event.getRequestId() : UUID.randomUUID());
+ this.enrichment = (event.getAai() != null ? event.getAai() : Map.of());
}
/**
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/ActorImpl.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/ActorImpl.java
index 9b9aa914e..d7f322e8a 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/ActorImpl.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/ActorImpl.java
@@ -20,14 +20,12 @@
package org.onap.policy.controlloop.actorserviceprovider.impl;
-import com.google.common.collect.ImmutableMap;
-import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import org.onap.policy.common.parameters.BeanValidationResult;
import org.onap.policy.controlloop.actorserviceprovider.Operator;
@@ -46,44 +44,42 @@ public class ActorImpl extends StartConfigPartial<Map<String, Object>> implement
/**
* Maps a name to an operator.
*/
- private Map<String, Operator> name2operator;
+ private final Map<String, Operator> name2operator = new ConcurrentHashMap<>();
/**
* Constructs the object.
*
* @param name actor name
- * @param operators the operations supported by this actor
*/
- public ActorImpl(String name, Operator... operators) {
+ public ActorImpl(String name) {
super(name);
- setOperators(Arrays.asList(operators));
}
/**
- * Sets the operators supported by this actor, overriding any previous list.
+ * Adds an operator supported by this actor.
*
- * @param operators the operations supported by this actor
+ * @param operator operation to be added
*/
- protected void setOperators(List<Operator> operators) {
+ protected synchronized void addOperator(Operator operator) {
+ /*
+ * This method is "synchronized" to prevent the state from changing while the
+ * operator is added. The map, itself, does not need synchronization as it's a
+ * concurrent map.
+ */
+
if (isConfigured()) {
throw new IllegalStateException("attempt to set operators on a configured actor: " + getName());
}
- Map<String, Operator> map = new HashMap<>();
- for (Operator newOp : operators) {
- map.compute(newOp.getName(), (opName, existingOp) -> {
- if (existingOp == null) {
- return newOp;
- }
-
- // TODO: should this throw an exception?
- logger.warn("duplicate names for actor operation {}.{}: {}, ignoring {}", getName(), opName,
- existingOp.getClass().getSimpleName(), newOp.getClass().getSimpleName());
- return existingOp;
- });
- }
+ name2operator.compute(operator.getName(), (opName, existingOp) -> {
+ if (existingOp == null) {
+ return operator;
+ }
- this.name2operator = ImmutableMap.copyOf(map);
+ logger.warn("duplicate names for actor operation {}.{}: {}, ignoring {}", getName(), opName,
+ existingOp.getClass().getSimpleName(), operator.getClass().getSimpleName());
+ return existingOp;
+ });
}
@Override
@@ -177,7 +173,7 @@ public class ActorImpl extends StartConfigPartial<Map<String, Object>> implement
for (Operator oper : name2operator.values()) {
if (oper.isConfigured()) {
- Util.logException(oper::start, "failed to start operation {}.{}", actorName, oper.getName());
+ Util.runFunction(oper::start, "failed to start operation {}.{}", actorName, oper.getName());
} else {
logger.warn("not starting unconfigured operation {}.{}", actorName, oper.getName());
@@ -195,7 +191,7 @@ public class ActorImpl extends StartConfigPartial<Map<String, Object>> implement
// @formatter:off
name2operator.values().forEach(
- oper -> Util.logException(oper::stop, "failed to stop operation {}.{}", actorName, oper.getName()));
+ oper -> Util.runFunction(oper::stop, "failed to stop operation {}.{}", actorName, oper.getName()));
// @formatter:on
}
@@ -208,7 +204,7 @@ public class ActorImpl extends StartConfigPartial<Map<String, Object>> implement
logger.info("shutting down operations for actor {}", actorName);
// @formatter:off
- name2operator.values().forEach(oper -> Util.logException(oper::shutdown,
+ name2operator.values().forEach(oper -> Util.runFunction(oper::shutdown,
"failed to shutdown operation {}.{}", actorName, oper.getName()));
// @formatter:on
}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpActor.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpActor.java
new file mode 100644
index 000000000..28b7b3924
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpActor.java
@@ -0,0 +1,58 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2020 AT&T Intellectual Property. 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actorserviceprovider.impl;
+
+import java.util.Map;
+import java.util.function.Function;
+import org.onap.policy.controlloop.actorserviceprovider.Util;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.HttpActorParams;
+
+/**
+ * Actor that uses HTTP, where the only additional property that an operator needs is a
+ * URL. The actor's parameters must be an {@link HttpActorParams} and its operator
+ * parameters are expected to be an {@link HttpParams}.
+ */
+public class HttpActor extends ActorImpl {
+
+ /**
+ * Constructs the object.
+ *
+ * @param name actor's name
+ */
+ public HttpActor(String name) {
+ super(name);
+ }
+
+ /**
+ * Translates the parameters to an {@link HttpActorParams} and then creates a function
+ * that will extract operator-specific parameters.
+ */
+ @Override
+ protected Function<String, Map<String, Object>> makeOperatorParameters(Map<String, Object> actorParameters) {
+ String actorName = getName();
+
+ // @formatter:off
+ return Util.translate(actorName, actorParameters, HttpActorParams.class)
+ .doValidation(actorName)
+ .makeOperationParameters(actorName);
+ // @formatter:on
+ }
+}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperator.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperator.java
new file mode 100644
index 000000000..566492907
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperator.java
@@ -0,0 +1,84 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2020 AT&T Intellectual Property. 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actorserviceprovider.impl;
+
+import java.util.Map;
+import lombok.AccessLevel;
+import lombok.Getter;
+import org.onap.policy.common.endpoints.http.client.HttpClient;
+import org.onap.policy.common.endpoints.http.client.HttpClientFactory;
+import org.onap.policy.common.endpoints.http.client.HttpClientFactoryInstance;
+import org.onap.policy.common.parameters.ValidationResult;
+import org.onap.policy.controlloop.actorserviceprovider.Util;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.HttpParams;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ParameterValidationRuntimeException;
+
+/**
+ * Operator that uses HTTP. The operator's parameters must be a {@link HttpParams}.
+ */
+public class HttpOperator extends OperatorPartial {
+
+ @Getter(AccessLevel.PROTECTED)
+ private HttpClient client;
+
+ @Getter
+ private long timeoutSec;
+
+ /**
+ * URI path for this particular operation.
+ */
+ @Getter
+ private String path;
+
+
+ /**
+ * Constructs the object.
+ *
+ * @param actorName name of the actor with which this operator is associated
+ * @param name operation name
+ */
+ public HttpOperator(String actorName, String name) {
+ super(actorName, name);
+ }
+
+ /**
+ * Translates the parameters to an {@link HttpParams} and then extracts the relevant
+ * values.
+ */
+ @Override
+ protected void doConfigure(Map<String, Object> parameters) {
+ HttpParams params = Util.translate(getFullName(), parameters, HttpParams.class);
+ ValidationResult result = params.validate(getFullName());
+ if (!result.isValid()) {
+ throw new ParameterValidationRuntimeException("invalid parameters", result);
+ }
+
+ client = getClientFactory().get(params.getClientName());
+ path = params.getPath();
+ timeoutSec = params.getTimeoutSec();
+ }
+
+ // these may be overridden by junits
+
+ protected HttpClientFactory getClientFactory() {
+ return HttpClientFactoryInstance.getClientFactory();
+ }
+}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperatorPartial.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperatorPartial.java
index 80d8fbd04..df5258d71 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperatorPartial.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperatorPartial.java
@@ -20,37 +20,61 @@
package org.onap.policy.controlloop.actorserviceprovider.impl;
-import java.time.Instant;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiConsumer;
import java.util.function.Function;
+import lombok.AccessLevel;
import lombok.Getter;
+import lombok.Setter;
import org.onap.policy.controlloop.ControlLoopOperation;
+import org.onap.policy.controlloop.actorserviceprovider.CallbackManager;
+import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
import org.onap.policy.controlloop.actorserviceprovider.Operator;
import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
import org.onap.policy.controlloop.actorserviceprovider.pipeline.PipelineControllerFuture;
-import org.onap.policy.controlloop.policy.Policy;
import org.onap.policy.controlloop.policy.PolicyResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * Partial implementation of an operator. Subclasses can choose to simply implement
- * {@link #doOperation(ControlLoopOperationParams)}, or they may choose to override
- * {@link #doOperationAsFuture(ControlLoopOperationParams)}.
+ * Partial implementation of an operator. In general, it's preferable that subclasses
+ * would override
+ * {@link #startOperationAsync(ControlLoopOperationParams, int, OperationOutcome)
+ * startOperationAsync()}. However, if that proves to be too difficult, then they can
+ * simply override {@link #doOperation(ControlLoopOperationParams, int, OperationOutcome)
+ * doOperation()}. In addition, if the operation requires any preprocessor steps, the
+ * subclass may choose to override
+ * {@link #startPreprocessorAsync(ControlLoopOperationParams) startPreprocessorAsync()}.
+ * <p/>
+ * The futures returned by the methods within this class can be canceled, and will
+ * propagate the cancellation to any subtasks. Thus it is also expected that any futures
+ * returned by overridden methods will do the same. Of course, if a class overrides
+ * {@link #doOperation(ControlLoopOperationParams, int, OperationOutcome) doOperation()},
+ * then there's little that can be done to cancel that particular operation.
*/
public abstract class OperatorPartial extends StartConfigPartial<Map<String, Object>> implements Operator {
private static final Logger logger = LoggerFactory.getLogger(OperatorPartial.class);
- private static final String OUTCOME_SUCCESS = PolicyResult.SUCCESS.toString();
- private static final String OUTCOME_FAILURE = PolicyResult.FAILURE.toString();
- private static final String OUTCOME_RETRIES = PolicyResult.FAILURE_RETRIES.toString();
+ /**
+ * Executor to be used for tasks that may perform blocking I/O. The default executor
+ * simply launches a new thread for each command that is submitted to it.
+ * <p/>
+ * May be overridden by junit tests.
+ */
+ @Getter(AccessLevel.PROTECTED)
+ @Setter(AccessLevel.PROTECTED)
+ private Executor blockingExecutor = command -> {
+ Thread thread = new Thread(command);
+ thread.setDaemon(true);
+ thread.start();
+ };
@Getter
private final String actorName;
@@ -103,94 +127,52 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
}
@Override
- public final CompletableFuture<ControlLoopOperation> startOperation(ControlLoopOperationParams params) {
+ public final CompletableFuture<OperationOutcome> startOperation(ControlLoopOperationParams params) {
if (!isAlive()) {
throw new IllegalStateException("operation is not running: " + getFullName());
}
- final Executor executor = params.getExecutor();
-
// allocate a controller for the entire operation
- final PipelineControllerFuture<ControlLoopOperation> controller = new PipelineControllerFuture<>();
+ final PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
- CompletableFuture<ControlLoopOperation> preproc = startPreprocessor(params);
+ CompletableFuture<OperationOutcome> preproc = startPreprocessorAsync(params);
if (preproc == null) {
// no preprocessor required - just start the operation
return startOperationAttempt(params, controller, 1);
}
- // propagate "stop" to the preprocessor
- controller.add(preproc);
-
/*
* Do preprocessor first and then, if successful, start the operation. Note:
* operations create their own outcome, ignoring the outcome from any previous
* steps.
+ *
+ * Wrap the preprocessor to ensure "stop" is propagated to it.
*/
- preproc.whenCompleteAsync(controller.delayedRemove(preproc), executor)
- .thenComposeAsync(handleFailure(params, controller), executor)
- .thenComposeAsync(onSuccess(params, unused -> startOperationAttempt(params, controller, 1)),
- executor);
-
- return controller;
- }
-
- /**
- * Starts an operation's preprocessor step(s). If the preprocessor fails, then it
- * invokes the started and completed call-backs.
- *
- * @param params operation parameters
- * @return a future that will return the preprocessor outcome, or {@code null} if this
- * operation needs no preprocessor
- */
- protected CompletableFuture<ControlLoopOperation> startPreprocessor(ControlLoopOperationParams params) {
- logger.info("{}: start low-level operation preprocessor for {}", getFullName(), params.getRequestId());
-
- final Executor executor = params.getExecutor();
- final ControlLoopOperation operation = params.makeOutcome();
-
- final Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> preproc =
- doPreprocessorAsFuture(params);
- if (preproc == null) {
- // no preprocessor required
- return null;
- }
-
- // allocate a controller for the preprocessor steps
- final PipelineControllerFuture<ControlLoopOperation> controller = new PipelineControllerFuture<>();
-
- /*
- * Don't mark it complete until we've built the whole pipeline. This will prevent
- * the operation from starting until after it has been successfully built (i.e.,
- * without generating any exceptions).
- */
- final CompletableFuture<ControlLoopOperation> firstFuture = new CompletableFuture<>();
-
// @formatter:off
- firstFuture
- .thenComposeAsync(controller.add(preproc), executor)
- .exceptionally(fromException(params, operation))
- .whenCompleteAsync(controller.delayedComplete(), executor);
+ controller.wrap(preproc)
+ .exceptionally(fromException(params, "preprocessor of operation"))
+ .thenCompose(handlePreprocessorFailure(params, controller))
+ .thenCompose(unusedOutcome -> startOperationAttempt(params, controller, 1));
// @formatter:on
- // start the pipeline
- firstFuture.complete(operation);
-
return controller;
}
/**
* Handles a failure in the preprocessor pipeline. If a failure occurred, then it
- * invokes the call-backs and returns a failed outcome. Otherwise, it returns the
- * outcome that it received.
+ * invokes the call-backs, marks the controller complete, and returns an incomplete
+ * future, effectively halting the pipeline. Otherwise, it returns the outcome that it
+ * received.
+ * <p/>
+ * Assumes that no callbacks have been invoked yet.
*
* @param params operation parameters
* @param controller pipeline controller
* @return a function that checks the outcome status and continues, if successful, or
* indicates a failure otherwise
*/
- private Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> handleFailure(
- ControlLoopOperationParams params, PipelineControllerFuture<ControlLoopOperation> controller) {
+ private Function<OperationOutcome, CompletableFuture<OperationOutcome>> handlePreprocessorFailure(
+ ControlLoopOperationParams params, PipelineControllerFuture<OperationOutcome> controller) {
return outcome -> {
@@ -207,19 +189,21 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
// propagate "stop" to the callbacks
controller.add(callbacks);
- final ControlLoopOperation outcome2 = params.makeOutcome();
+ final OperationOutcome outcome2 = params.makeOutcome();
// TODO need a FAILURE_MISSING_DATA (e.g., A&AI)
- outcome2.setOutcome(PolicyResult.FAILURE_GUARD.toString());
+ outcome2.setResult(PolicyResult.FAILURE_GUARD);
outcome2.setMessage(outcome != null ? outcome.getMessage() : null);
- CompletableFuture.completedFuture(outcome2).thenApplyAsync(callbackStarted(params, callbacks), executor)
- .thenApplyAsync(callbackCompleted(params, callbacks), executor)
- .whenCompleteAsync(controller.delayedRemove(callbacks), executor)
+ // @formatter:off
+ CompletableFuture.completedFuture(outcome2)
+ .whenCompleteAsync(callbackStarted(params, callbacks), executor)
+ .whenCompleteAsync(callbackCompleted(params, callbacks), executor)
.whenCompleteAsync(controller.delayedComplete(), executor);
+ // @formatter:on
- return controller;
+ return new CompletableFuture<>();
};
}
@@ -237,8 +221,7 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
* @return a function that will start the preprocessor and returns its outcome, or
* {@code null} if this operation needs no preprocessor
*/
- protected Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> doPreprocessorAsFuture(
- ControlLoopOperationParams params) {
+ protected CompletableFuture<OperationOutcome> startPreprocessorAsync(ControlLoopOperationParams params) {
return null;
}
@@ -251,20 +234,12 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
* @param attempt attempt number, typically starting with 1
* @return a future that will return the final result of all attempts
*/
- private CompletableFuture<ControlLoopOperation> startOperationAttempt(ControlLoopOperationParams params,
- PipelineControllerFuture<ControlLoopOperation> controller, int attempt) {
-
- final Executor executor = params.getExecutor();
-
- CompletableFuture<ControlLoopOperation> future = startAttemptWithoutRetries(params, attempt);
+ private CompletableFuture<OperationOutcome> startOperationAttempt(ControlLoopOperationParams params,
+ PipelineControllerFuture<OperationOutcome> controller, int attempt) {
// propagate "stop" to the operation attempt
- controller.add(future);
-
- // detach when complete
- future.whenCompleteAsync(controller.delayedRemove(future), executor)
- .thenComposeAsync(retryOnFailure(params, controller, attempt), params.getExecutor())
- .whenCompleteAsync(controller.delayedComplete(), executor);
+ controller.wrap(startAttemptWithoutRetries(params, attempt))
+ .thenCompose(retryOnFailure(params, controller, attempt));
return controller;
}
@@ -276,40 +251,32 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
* @param attempt attempt number, typically starting with 1
* @return a future that will return the result of a single operation attempt
*/
- private CompletableFuture<ControlLoopOperation> startAttemptWithoutRetries(ControlLoopOperationParams params,
+ private CompletableFuture<OperationOutcome> startAttemptWithoutRetries(ControlLoopOperationParams params,
int attempt) {
logger.info("{}: start operation attempt {} for {}", getFullName(), attempt, params.getRequestId());
final Executor executor = params.getExecutor();
- final ControlLoopOperation outcome = params.makeOutcome();
+ final OperationOutcome outcome = params.makeOutcome();
final CallbackManager callbacks = new CallbackManager();
// this operation attempt gets its own controller
- final PipelineControllerFuture<ControlLoopOperation> controller = new PipelineControllerFuture<>();
+ final PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
// propagate "stop" to the callbacks
controller.add(callbacks);
- /*
- * Don't mark it complete until we've built the whole pipeline. This will prevent
- * the operation from starting until after it has been successfully built (i.e.,
- * without generating any exceptions).
- */
- final CompletableFuture<ControlLoopOperation> firstFuture = new CompletableFuture<>();
-
// @formatter:off
- CompletableFuture<ControlLoopOperation> future2 =
- firstFuture.thenComposeAsync(verifyRunning(controller, params), executor)
- .thenApplyAsync(callbackStarted(params, callbacks), executor)
- .thenComposeAsync(controller.add(doOperationAsFuture(params, attempt)), executor);
+ CompletableFuture<OperationOutcome> future = CompletableFuture.completedFuture(outcome)
+ .whenCompleteAsync(callbackStarted(params, callbacks), executor)
+ .thenCompose(controller.wrap(outcome2 -> startOperationAsync(params, attempt, outcome2)));
// @formatter:on
// handle timeouts, if specified
- long timeoutMillis = getTimeOutMillis(params.getPolicy());
+ long timeoutMillis = getTimeOutMillis(params.getTimeoutSec());
if (timeoutMillis > 0) {
logger.info("{}: set timeout to {}ms for {}", getFullName(), timeoutMillis, params.getRequestId());
- future2 = future2.orTimeout(timeoutMillis, TimeUnit.MILLISECONDS);
+ future = future.orTimeout(timeoutMillis, TimeUnit.MILLISECONDS);
}
/*
@@ -321,16 +288,13 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
*/
// @formatter:off
- future2.exceptionally(fromException(params, outcome))
- .thenApplyAsync(setRetryFlag(params, attempt), executor)
- .thenApplyAsync(callbackStarted(params, callbacks), executor)
- .thenApplyAsync(callbackCompleted(params, callbacks), executor)
+ future.exceptionally(fromException(params, "operation"))
+ .thenApply(setRetryFlag(params, attempt))
+ .whenCompleteAsync(callbackStarted(params, callbacks), executor)
+ .whenCompleteAsync(callbackCompleted(params, callbacks), executor)
.whenCompleteAsync(controller.delayedComplete(), executor);
// @formatter:on
- // start the pipeline
- firstFuture.complete(outcome);
-
return controller;
}
@@ -340,8 +304,8 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
* @param outcome outcome to examine
* @return {@code true} if the outcome was successful
*/
- protected boolean isSuccess(ControlLoopOperation outcome) {
- return OUTCOME_SUCCESS.equals(outcome.getOutcome());
+ protected boolean isSuccess(OperationOutcome outcome) {
+ return (outcome.getResult() == PolicyResult.SUCCESS);
}
/**
@@ -351,13 +315,29 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
* @return {@code true} if the outcome is not {@code null} and was a failure
* <i>and</i> was associated with this operator, {@code false} otherwise
*/
- protected boolean isActorFailed(ControlLoopOperation outcome) {
- return OUTCOME_FAILURE.equals(getActorOutcome(outcome));
+ protected boolean isActorFailed(OperationOutcome outcome) {
+ return (isSameOperation(outcome) && outcome.getResult() == PolicyResult.FAILURE);
+ }
+
+ /**
+ * Determines if the given outcome is for this operation.
+ *
+ * @param outcome outcome to examine
+ * @return {@code true} if the outcome is for this operation, {@code false} otherwise
+ */
+ protected boolean isSameOperation(OperationOutcome outcome) {
+ return OperationOutcome.isFor(outcome, getActorName(), getName());
}
/**
* Invokes the operation as a "future". This method simply invokes
- * {@link #doOperation(ControlLoopOperationParams)} turning it into a "future".
+ * {@link #doOperation(ControlLoopOperationParams)} using the {@link #blockingExecutor
+ * "blocking executor"}, returning the result via a "future".
+ * <p/>
+ * Note: if the operation uses blocking I/O, then it should <i>not</i> be run using
+ * the executor in the "params", as that may bring the background thread pool to a
+ * grinding halt. The {@link #blockingExecutor "blocking executor"} should be used
+ * instead.
* <p/>
* This method assumes the following:
* <ul>
@@ -373,31 +353,23 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
* @return a function that will start the operation and return its result when
* complete
*/
- protected Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> doOperationAsFuture(
- ControlLoopOperationParams params, int attempt) {
-
- /*
- * TODO As doOperation() may perform blocking I/O, this should be launched in its
- * own thread to prevent the ForkJoinPool from being tied up. Should probably
- * provide a method to make that easy.
- */
+ protected CompletableFuture<OperationOutcome> startOperationAsync(ControlLoopOperationParams params, int attempt,
+ OperationOutcome outcome) {
- return operation -> CompletableFuture.supplyAsync(() -> doOperation(params, attempt, operation),
- params.getExecutor());
+ return CompletableFuture.supplyAsync(() -> doOperation(params, attempt, outcome), getBlockingExecutor());
}
/**
* Low-level method that performs the operation. This can make the same assumptions
* that are made by {@link #doOperationAsFuture(ControlLoopOperationParams)}. This
- * method throws an {@link UnsupportedOperationException}.
+ * particular method simply throws an {@link UnsupportedOperationException}.
*
* @param params operation parameters
* @param attempt attempt number, typically starting with 1
* @param operation the operation being performed
* @return the outcome of the operation
*/
- protected ControlLoopOperation doOperation(ControlLoopOperationParams params, int attempt,
- ControlLoopOperation operation) {
+ protected OperationOutcome doOperation(ControlLoopOperationParams params, int attempt, OperationOutcome operation) {
throw new UnsupportedOperationException("start operation " + getFullName());
}
@@ -411,8 +383,7 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
* @param attempt latest attempt number, starting with 1
* @return a function to get the next future to execute
*/
- private Function<ControlLoopOperation, ControlLoopOperation> setRetryFlag(ControlLoopOperationParams params,
- int attempt) {
+ private Function<OperationOutcome, OperationOutcome> setRetryFlag(ControlLoopOperationParams params, int attempt) {
return operation -> {
if (operation != null && !isActorFailed(operation)) {
@@ -424,22 +395,22 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
}
// get a non-null operation
- ControlLoopOperation oper2;
+ OperationOutcome oper2;
if (operation != null) {
oper2 = operation;
} else {
oper2 = params.makeOutcome();
- oper2.setOutcome(OUTCOME_FAILURE);
+ oper2.setResult(PolicyResult.FAILURE);
}
- if (params.getPolicy().getRetry() != null && params.getPolicy().getRetry() > 0
- && attempt > params.getPolicy().getRetry()) {
+ Integer retry = params.getRetry();
+ if (retry != null && retry > 0 && attempt > retry) {
/*
* retries were specified and we've already tried them all - change to
* FAILURE_RETRIES
*/
logger.info("operation {} retries exhausted for {}", getFullName(), params.getRequestId());
- oper2.setOutcome(OUTCOME_RETRIES);
+ oper2.setResult(PolicyResult.FAILURE_RETRIES);
}
return oper2;
@@ -456,21 +427,24 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
* @param attempt latest attempt number, starting with 1
* @return a function to get the next future to execute
*/
- private Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> retryOnFailure(
- ControlLoopOperationParams params, PipelineControllerFuture<ControlLoopOperation> controller,
+ private Function<OperationOutcome, CompletableFuture<OperationOutcome>> retryOnFailure(
+ ControlLoopOperationParams params, PipelineControllerFuture<OperationOutcome> controller,
int attempt) {
return operation -> {
if (!isActorFailed(operation)) {
// wrong type or wrong operation - just leave it as is
logger.trace("not retrying operation {} for {}", getFullName(), params.getRequestId());
- return CompletableFuture.completedFuture(operation);
+ controller.complete(operation);
+ return new CompletableFuture<>();
}
- if (params.getPolicy().getRetry() == null || params.getPolicy().getRetry() <= 0) {
+ Integer retry = params.getRetry();
+ if (retry == null || retry <= 0) {
// no retries - already marked as FAILURE, so just return it
logger.info("operation {} no retries for {}", getFullName(), params.getRequestId());
- return CompletableFuture.completedFuture(operation);
+ controller.complete(operation);
+ return new CompletableFuture<>();
}
@@ -484,100 +458,279 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
}
/**
- * Gets the outcome of an operation for this operation.
+ * Converts an exception into an operation outcome, returning a copy of the outcome to
+ * prevent background jobs from changing it.
*
- * @param operation operation whose outcome is to be extracted
- * @return the outcome of the given operation, if it's for this operator, {@code null}
- * otherwise
+ * @param params operation parameters
+ * @param type type of item throwing the exception
+ * @return a function that will convert an exception into an operation outcome
*/
- protected String getActorOutcome(ControlLoopOperation operation) {
- if (operation == null) {
- return null;
- }
+ private Function<Throwable, OperationOutcome> fromException(ControlLoopOperationParams params, String type) {
- if (!getActorName().equals(operation.getActor())) {
- return null;
- }
+ return thrown -> {
+ OperationOutcome outcome = params.makeOutcome();
+
+ logger.warn("exception throw by {} {}.{} for {}", type, outcome.getActor(), outcome.getOperation(),
+ params.getRequestId(), thrown);
+
+ return setOutcome(params, outcome, thrown);
+ };
+ }
+
+ /**
+ * Similar to {@link CompletableFuture#anyOf(CompletableFuture...)}, but it cancels
+ * any outstanding futures when one completes.
+ *
+ * @param params operation parameters
+ * @param futures futures for which to wait
+ * @return a future to cancel or await an outcome. If this future is canceled, then
+ * all of the futures will be canceled
+ */
+ protected CompletableFuture<OperationOutcome> anyOf(ControlLoopOperationParams params,
+ List<CompletableFuture<OperationOutcome>> futures) {
+
+ // convert list to an array
+ @SuppressWarnings("rawtypes")
+ CompletableFuture[] arrFutures = futures.toArray(new CompletableFuture[futures.size()]);
+
+ @SuppressWarnings("unchecked")
+ CompletableFuture<OperationOutcome> result = anyOf(params, arrFutures);
+ return result;
+ }
+
+ /**
+ * Same as {@link CompletableFuture#anyOf(CompletableFuture...)}, but it cancels any
+ * outstanding futures when one completes.
+ *
+ * @param params operation parameters
+ * @param futures futures for which to wait
+ * @return a future to cancel or await an outcome. If this future is canceled, then
+ * all of the futures will be canceled
+ */
+ protected CompletableFuture<OperationOutcome> anyOf(ControlLoopOperationParams params,
+ @SuppressWarnings("unchecked") CompletableFuture<OperationOutcome>... futures) {
+
+ final Executor executor = params.getExecutor();
+ final PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+
+ attachFutures(controller, futures);
+
+ // @formatter:off
+ CompletableFuture.anyOf(futures)
+ .thenApply(object -> (OperationOutcome) object)
+ .whenCompleteAsync(controller.delayedComplete(), executor);
+ // @formatter:on
+
+ return controller;
+ }
+
+ /**
+ * Similar to {@link CompletableFuture#allOf(CompletableFuture...)}, but it cancels
+ * the futures if returned future is canceled. The future returns the "worst" outcome,
+ * based on priority (see {@link #detmPriority(OperationOutcome)}).
+ *
+ * @param params operation parameters
+ * @param futures futures for which to wait
+ * @return a future to cancel or await an outcome. If this future is canceled, then
+ * all of the futures will be canceled
+ */
+ protected CompletableFuture<OperationOutcome> allOf(ControlLoopOperationParams params,
+ List<CompletableFuture<OperationOutcome>> futures) {
- if (!getName().equals(operation.getOperation())) {
- return null;
+ // convert list to an array
+ @SuppressWarnings("rawtypes")
+ CompletableFuture[] arrFutures = futures.toArray(new CompletableFuture[futures.size()]);
+
+ @SuppressWarnings("unchecked")
+ CompletableFuture<OperationOutcome> result = allOf(params, arrFutures);
+ return result;
+ }
+
+ /**
+ * Same as {@link CompletableFuture#allOf(CompletableFuture...)}, but it cancels the
+ * futures if returned future is canceled. The future returns the "worst" outcome,
+ * based on priority (see {@link #detmPriority(OperationOutcome)}).
+ *
+ * @param params operation parameters
+ * @param futures futures for which to wait
+ * @return a future to cancel or await an outcome. If this future is canceled, then
+ * all of the futures will be canceled
+ */
+ protected CompletableFuture<OperationOutcome> allOf(ControlLoopOperationParams params,
+ @SuppressWarnings("unchecked") CompletableFuture<OperationOutcome>... futures) {
+
+ final PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+
+ attachFutures(controller, futures);
+
+ OperationOutcome[] outcomes = new OperationOutcome[futures.length];
+
+ @SuppressWarnings("rawtypes")
+ CompletableFuture[] futures2 = new CompletableFuture[futures.length];
+
+ // record the outcomes of each future when it completes
+ for (int count = 0; count < futures2.length; ++count) {
+ final int count2 = count;
+ futures2[count] = futures[count].whenComplete((outcome2, thrown) -> outcomes[count2] = outcome2);
}
- return operation.getOutcome();
+ CompletableFuture.allOf(futures2).whenComplete(combineOutcomes(params, controller, outcomes));
+
+ return controller;
}
/**
- * Gets a function that will start the next step, if the current operation was
- * successful, or just return the current operation, otherwise.
+ * Attaches the given futures to the controller.
+ *
+ * @param controller master controller for all of the futures
+ * @param futures futures to be attached to the controller
+ */
+ private void attachFutures(PipelineControllerFuture<OperationOutcome> controller,
+ @SuppressWarnings("unchecked") CompletableFuture<OperationOutcome>... futures) {
+
+ // attach each task
+ for (CompletableFuture<OperationOutcome> future : futures) {
+ controller.add(future);
+ }
+ }
+
+ /**
+ * Combines the outcomes from a set of tasks.
*
* @param params operation parameters
- * @param nextStep function that will invoke the next step, passing it the operation
- * @return a function that will start the next step
+ * @param future future to be completed with the combined result
+ * @param outcomes outcomes to be examined
*/
- protected Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> onSuccess(
- ControlLoopOperationParams params,
- Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> nextStep) {
+ private BiConsumer<Void, Throwable> combineOutcomes(ControlLoopOperationParams params,
+ CompletableFuture<OperationOutcome> future, OperationOutcome[] outcomes) {
- return operation -> {
+ return (unused, thrown) -> {
+ if (thrown != null) {
+ future.completeExceptionally(thrown);
+ return;
+ }
- if (operation == null) {
- logger.trace("{}: null outcome - discarding next task for {}", getFullName(), params.getRequestId());
- ControlLoopOperation outcome = params.makeOutcome();
- outcome.setOutcome(OUTCOME_FAILURE);
- return CompletableFuture.completedFuture(outcome);
+ // identify the outcome with the highest priority
+ OperationOutcome outcome = outcomes[0];
+ int priority = detmPriority(outcome);
- } else if (isSuccess(operation)) {
- logger.trace("{}: success - starting next task for {}", getFullName(), params.getRequestId());
- return nextStep.apply(operation);
+ // start with "1", as we've already dealt with "0"
+ for (int count = 1; count < outcomes.length; ++count) {
+ OperationOutcome outcome2 = outcomes[count];
+ int priority2 = detmPriority(outcome2);
- } else {
- logger.trace("{}: failure - discarding next task for {}", getFullName(), params.getRequestId());
- return CompletableFuture.completedFuture(operation);
+ if (priority2 > priority) {
+ outcome = outcome2;
+ priority = priority2;
+ }
}
+
+ logger.trace("{}: combined outcome of tasks is {} for {}", getFullName(),
+ (outcome == null ? null : outcome.getResult()), params.getRequestId());
+
+ future.complete(outcome);
};
}
/**
- * Converts an exception into an operation outcome, returning a copy of the outcome to
- * prevent background jobs from changing it.
+ * Determines the priority of an outcome based on its result.
*
- * @param params operation parameters
- * @param operation current operation
- * @return a function that will convert an exception into an operation outcome
+ * @param outcome outcome to examine, or {@code null}
+ * @return the outcome's priority
*/
- private Function<Throwable, ControlLoopOperation> fromException(ControlLoopOperationParams params,
- ControlLoopOperation operation) {
+ protected int detmPriority(OperationOutcome outcome) {
+ if (outcome == null) {
+ return 1;
+ }
- return thrown -> {
- logger.warn("exception throw by operation {}.{} for {}", operation.getActor(), operation.getOperation(),
- params.getRequestId(), thrown);
+ switch (outcome.getResult()) {
+ case SUCCESS:
+ return 0;
+
+ case FAILURE_GUARD:
+ return 2;
+
+ case FAILURE_RETRIES:
+ return 3;
+
+ case FAILURE:
+ return 4;
+
+ case FAILURE_TIMEOUT:
+ return 5;
+
+ case FAILURE_EXCEPTION:
+ default:
+ return 6;
+ }
+ }
+
+ /**
+ * Performs a task, after verifying that the controller is still running. Also checks
+ * that the previous outcome was successful, if specified.
+ *
+ * @param params operation parameters
+ * @param controller overall pipeline controller
+ * @param checkSuccess {@code true} to check the previous outcome, {@code false}
+ * otherwise
+ * @param outcome outcome of the previous task
+ * @param tasks tasks to be performed
+ * @return a function to perform the task. If everything checks out, then it returns
+ * the task's future. Otherwise, it returns an incomplete future and completes
+ * the controller instead.
+ */
+ // @formatter:off
+ protected CompletableFuture<OperationOutcome> doTask(ControlLoopOperationParams params,
+ PipelineControllerFuture<OperationOutcome> controller,
+ boolean checkSuccess, OperationOutcome outcome,
+ CompletableFuture<OperationOutcome> task) {
+ // @formatter:on
+ if (checkSuccess && !isSuccess(outcome)) {
/*
- * Must make a copy of the operation, as the original could be changed by
- * background jobs that might still be running.
+ * must complete before canceling so that cancel() doesn't cause controller to
+ * complete
*/
- return setOutcome(params, new ControlLoopOperation(operation), thrown);
- };
+ controller.complete(outcome);
+ task.cancel(false);
+ return new CompletableFuture<>();
+ }
+
+ return controller.wrap(task);
}
/**
- * Gets a function to verify that the operation is still running. If the pipeline is
- * not running, then it returns an incomplete future, which will effectively halt
- * subsequent operations in the pipeline. This method is intended to be used with one
- * of the {@link CompletableFuture}'s <i>thenCompose()</i> methods.
+ * Performs a task, after verifying that the controller is still running. Also checks
+ * that the previous outcome was successful, if specified.
*
- * @param controller pipeline controller
* @param params operation parameters
- * @return a function to verify that the operation is still running
+ * @param controller overall pipeline controller
+ * @param checkSuccess {@code true} to check the previous outcome, {@code false}
+ * otherwise
+ * @param tasks tasks to be performed
+ * @return a function to perform the task. If everything checks out, then it returns
+ * the task's future. Otherwise, it returns an incomplete future and completes
+ * the controller instead.
*/
- protected <T> Function<T, CompletableFuture<T>> verifyRunning(
- PipelineControllerFuture<ControlLoopOperation> controller, ControlLoopOperationParams params) {
+ // @formatter:off
+ protected Function<OperationOutcome, CompletableFuture<OperationOutcome>> doTask(ControlLoopOperationParams params,
+ PipelineControllerFuture<OperationOutcome> controller,
+ boolean checkSuccess,
+ Function<OperationOutcome, CompletableFuture<OperationOutcome>> task) {
+ // @formatter:on
+
+ return outcome -> {
+
+ if (!controller.isRunning()) {
+ return new CompletableFuture<>();
+ }
- return value -> {
- boolean running = controller.isRunning();
- logger.trace("{}: verify running {} for {}", getFullName(), running, params.getRequestId());
+ if (checkSuccess && !isSuccess(outcome)) {
+ controller.complete(outcome);
+ return new CompletableFuture<>();
+ }
- return (running ? CompletableFuture.completedFuture(value) : new CompletableFuture<>());
+ return controller.wrap(task.apply(outcome));
};
}
@@ -591,10 +744,10 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
* @param callbacks used to determine if the start callback can be invoked
* @return a function that sets the start time and invokes the callback
*/
- private Function<ControlLoopOperation, ControlLoopOperation> callbackStarted(ControlLoopOperationParams params,
+ private BiConsumer<OperationOutcome, Throwable> callbackStarted(ControlLoopOperationParams params,
CallbackManager callbacks) {
- return outcome -> {
+ return (outcome, thrown) -> {
if (callbacks.canStart()) {
// haven't invoked "start" callback yet
@@ -602,8 +755,6 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
outcome.setEnd(null);
params.callbackStarted(outcome);
}
-
- return outcome;
};
}
@@ -621,18 +772,16 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
* @param callbacks used to determine if the end callback can be invoked
* @return a function that sets the end time and invokes the callback
*/
- private Function<ControlLoopOperation, ControlLoopOperation> callbackCompleted(ControlLoopOperationParams params,
+ private BiConsumer<OperationOutcome, Throwable> callbackCompleted(ControlLoopOperationParams params,
CallbackManager callbacks) {
- return operation -> {
+ return (outcome, thrown) -> {
if (callbacks.canEnd()) {
- operation.setStart(callbacks.getStartTime());
- operation.setEnd(callbacks.getEndTime());
- params.callbackCompleted(operation);
+ outcome.setStart(callbacks.getStartTime());
+ outcome.setEnd(callbacks.getEndTime());
+ params.callbackCompleted(outcome);
}
-
- return operation;
};
}
@@ -643,7 +792,7 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
* @param operation operation to be updated
* @return the updated operation
*/
- protected ControlLoopOperation setOutcome(ControlLoopOperationParams params, ControlLoopOperation operation,
+ protected OperationOutcome setOutcome(ControlLoopOperationParams params, OperationOutcome operation,
Throwable thrown) {
PolicyResult result = (isTimeout(thrown) ? PolicyResult.FAILURE_TIMEOUT : PolicyResult.FAILURE_EXCEPTION);
return setOutcome(params, operation, result);
@@ -657,10 +806,10 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
* @param result result of the operation
* @return the updated operation
*/
- protected ControlLoopOperation setOutcome(ControlLoopOperationParams params, ControlLoopOperation operation,
+ protected OperationOutcome setOutcome(ControlLoopOperationParams params, OperationOutcome operation,
PolicyResult result) {
logger.trace("{}: set outcome {} for {}", getFullName(), result, params.getRequestId());
- operation.setOutcome(result.toString());
+ operation.setResult(result);
operation.setMessage(result == PolicyResult.SUCCESS ? ControlLoopOperation.SUCCESS_MSG
: ControlLoopOperation.FAILED_MSG);
@@ -687,71 +836,10 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
* Gets the operation timeout. Subclasses may override this method to obtain the
* timeout in some other way (e.g., through configuration properties).
*
- * @param policy policy from which to extract the timeout
+ * @param timeoutSec timeout, in seconds, or {@code null}
* @return the operation timeout, in milliseconds
*/
- protected long getTimeOutMillis(Policy policy) {
- Integer timeoutSec = policy.getTimeout();
+ protected long getTimeOutMillis(Integer timeoutSec) {
return (timeoutSec == null ? 0 : TimeUnit.MILLISECONDS.convert(timeoutSec, TimeUnit.SECONDS));
}
-
- /**
- * Manager for "start" and "end" callbacks.
- */
- private static class CallbackManager implements Runnable {
- private final AtomicReference<Instant> startTime = new AtomicReference<>();
- private final AtomicReference<Instant> endTime = new AtomicReference<>();
-
- /**
- * Determines if the "start" callback can be invoked. If so, it sets the
- * {@link #startTime} to the current time.
- *
- * @return {@code true} if the "start" callback can be invoked, {@code false}
- * otherwise
- */
- public boolean canStart() {
- return startTime.compareAndSet(null, Instant.now());
- }
-
- /**
- * Determines if the "end" callback can be invoked. If so, it sets the
- * {@link #endTime} to the current time.
- *
- * @return {@code true} if the "end" callback can be invoked, {@code false}
- * otherwise
- */
- public boolean canEnd() {
- return endTime.compareAndSet(null, Instant.now());
- }
-
- /**
- * Gets the start time.
- *
- * @return the start time, or {@code null} if {@link #canStart()} has not been
- * invoked yet.
- */
- public Instant getStartTime() {
- return startTime.get();
- }
-
- /**
- * Gets the end time.
- *
- * @return the end time, or {@code null} if {@link #canEnd()} has not been invoked
- * yet.
- */
- public Instant getEndTime() {
- return endTime.get();
- }
-
- /**
- * Prevents further callbacks from being executed by setting {@link #startTime}
- * and {@link #endTime}.
- */
- @Override
- public void run() {
- canStart();
- canEnd();
- }
- }
}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/parameters/ControlLoopOperationParams.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/parameters/ControlLoopOperationParams.java
index 08aba81f2..57fce40d7 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/parameters/ControlLoopOperationParams.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/parameters/ControlLoopOperationParams.java
@@ -20,6 +20,7 @@
package org.onap.policy.controlloop.actorserviceprovider.parameters;
+import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
@@ -32,11 +33,11 @@ import lombok.Getter;
import org.onap.policy.common.parameters.BeanValidationResult;
import org.onap.policy.common.parameters.BeanValidator;
import org.onap.policy.common.parameters.annotations.NotNull;
-import org.onap.policy.controlloop.ControlLoopOperation;
import org.onap.policy.controlloop.actorserviceprovider.ActorService;
+import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
import org.onap.policy.controlloop.actorserviceprovider.Util;
import org.onap.policy.controlloop.actorserviceprovider.controlloop.ControlLoopEventContext;
-import org.onap.policy.controlloop.policy.Policy;
+import org.onap.policy.controlloop.policy.Target;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,36 +50,67 @@ import org.slf4j.LoggerFactory;
@AllArgsConstructor
@EqualsAndHashCode
public class ControlLoopOperationParams {
-
private static final Logger logger = LoggerFactory.getLogger(ControlLoopOperationParams.class);
- public static final String UNKNOWN = "-unknown-";
-
+ /**
+ * Actor name.
+ */
+ @NotNull
+ private String actor;
/**
- * The actor service in which to find the actor/operation.
+ * Actor service in which to find the actor/operation.
*/
@NotNull
private ActorService actorService;
/**
- * The event for which the operation applies.
+ * Event for which the operation applies.
*/
@NotNull
private ControlLoopEventContext context;
/**
- * The executor to use to run the operation.
+ * Executor to use to run the operation.
*/
@NotNull
@Builder.Default
private Executor executor = ForkJoinPool.commonPool();
/**
- * The policy associated with the operation.
+ * Operation name.
+ */
+ @NotNull
+ private String operation;
+
+ /**
+ * Payload data for the request.
+ */
+ private Map<String, String> payload;
+
+ /**
+ * Number of retries allowed, or {@code null} if no retries.
+ */
+ private Integer retry;
+
+ /**
+ * The entity's target information. May be {@code null}, depending on the requirement
+ * of the operation to be invoked.
+ */
+ private Target target;
+
+ /**
+ * Target entity.
*/
@NotNull
- private Policy policy;
+ private String targetEntity;
+
+ /**
+ * Timeout, in seconds, or {@code null} if no timeout. Zero and negative values also
+ * imply no timeout.
+ */
+ @Builder.Default
+ private Integer timeoutSec = 300;
/**
* The function to invoke when the operation starts. This is optional.
@@ -87,7 +119,7 @@ public class ControlLoopOperationParams {
* may happen if the current operation requires other operations to be performed first
* (e.g., A&AI queries, guard checks).
*/
- private Consumer<ControlLoopOperation> startCallback;
+ private Consumer<OperationOutcome> startCallback;
/**
* The function to invoke when the operation completes. This is optional.
@@ -96,13 +128,7 @@ public class ControlLoopOperationParams {
* may happen if the current operation requires other operations to be performed first
* (e.g., A&AI queries, guard checks).
*/
- private Consumer<ControlLoopOperation> completeCallback;
-
- /**
- * Target entity.
- */
- @NotNull
- private String target;
+ private Consumer<OperationOutcome> completeCallback;
/**
* Starts the specified operation.
@@ -110,7 +136,7 @@ public class ControlLoopOperationParams {
* @return a future that will return the result of the operation
* @throws IllegalArgumentException if the parameters are invalid
*/
- public CompletableFuture<ControlLoopOperation> start() {
+ public CompletableFuture<OperationOutcome> start() {
BeanValidationResult result = validate();
if (!result.isValid()) {
logger.warn("parameter error in operation {}.{} for {}:\n{}", getActor(), getOperation(), getRequestId(),
@@ -120,31 +146,13 @@ public class ControlLoopOperationParams {
// @formatter:off
return actorService
- .getActor(policy.getActor())
- .getOperator(policy.getRecipe())
+ .getActor(getActor())
+ .getOperator(getOperation())
.startOperation(this);
// @formatter:on
}
/**
- * Gets the name of the actor from the policy.
- *
- * @return the actor name, or {@link #UNKNOWN} if no name is available
- */
- public String getActor() {
- return (policy == null || policy.getActor() == null ? UNKNOWN : policy.getActor());
- }
-
- /**
- * Gets the name of the operation from the policy.
- *
- * @return the operation name, or {@link #UNKNOWN} if no name is available
- */
- public String getOperation() {
- return (policy == null || policy.getRecipe() == null ? UNKNOWN : policy.getRecipe());
- }
-
- /**
* Gets the requested ID of the associated event.
*
* @return the event's request ID, or {@code null} if no request ID is available
@@ -158,13 +166,13 @@ public class ControlLoopOperationParams {
*
* @return a new operation outcome
*/
- public ControlLoopOperation makeOutcome() {
- ControlLoopOperation operation = new ControlLoopOperation();
- operation.setActor(getActor());
- operation.setOperation(getOperation());
- operation.setTarget(target);
+ public OperationOutcome makeOutcome() {
+ OperationOutcome outcome = new OperationOutcome();
+ outcome.setActor(getActor());
+ outcome.setOperation(getOperation());
+ outcome.setTarget(targetEntity);
- return operation;
+ return outcome;
}
/**
@@ -173,11 +181,11 @@ public class ControlLoopOperationParams {
*
* @param operation the operation that is being started
*/
- public void callbackStarted(ControlLoopOperation operation) {
+ public void callbackStarted(OperationOutcome operation) {
logger.info("started operation {}.{} for {}", operation.getActor(), operation.getOperation(), getRequestId());
if (startCallback != null) {
- Util.logException(() -> startCallback.accept(operation), "{}.{}: start-callback threw an exception for {}",
+ Util.runFunction(() -> startCallback.accept(operation), "{}.{}: start-callback threw an exception for {}",
operation.getActor(), operation.getOperation(), getRequestId());
}
}
@@ -188,12 +196,12 @@ public class ControlLoopOperationParams {
*
* @param operation the operation that is being started
*/
- public void callbackCompleted(ControlLoopOperation operation) {
+ public void callbackCompleted(OperationOutcome operation) {
logger.info("completed operation {}.{} outcome={} for {}", operation.getActor(), operation.getOperation(),
- operation.getOutcome(), getRequestId());
+ operation.getResult(), getRequestId());
if (completeCallback != null) {
- Util.logException(() -> completeCallback.accept(operation),
+ Util.runFunction(() -> completeCallback.accept(operation),
"{}.{}: complete-callback threw an exception for {}", operation.getActor(),
operation.getOperation(), getRequestId());
}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/ListenerManager.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/ListenerManager.java
index d34a3fb5b..1d64a8710 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/ListenerManager.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/ListenerManager.java
@@ -101,7 +101,7 @@ public class ListenerManager {
*/
protected void runListener(Runnable listener) {
// TODO do this asynchronously?
- Util.logException(listener, "pipeline listener {} threw an exception", listener);
+ Util.runFunction(listener, "pipeline listener {} threw an exception", listener);
}
/**
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/PipelineControllerFuture.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/PipelineControllerFuture.java
index 96c8f9e05..92843e28a 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/PipelineControllerFuture.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/PipelineControllerFuture.java
@@ -23,41 +23,70 @@ package org.onap.policy.controlloop.actorserviceprovider.pipeline;
import static org.onap.policy.controlloop.actorserviceprovider.Util.ident;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.Function;
+import java.util.function.Supplier;
import lombok.NoArgsConstructor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Pipeline controller, used by operations within the pipeline to determine if they should
- * continue to run. If {@link #cancel(boolean)} is invoked, it automatically stops the
- * pipeline.
+ * continue to run. Whenever this is canceled or completed, it automatically cancels all
+ * futures and runs all listeners that have been added.
*/
@NoArgsConstructor
public class PipelineControllerFuture<T> extends CompletableFuture<T> {
private static final Logger logger = LoggerFactory.getLogger(PipelineControllerFuture.class);
+ private static final String COMPLETE_EXCEPT_MSG = "{}: complete future with exception";
+ private static final String CANCEL_MSG = "{}: cancel future";
+ private static final String COMPLETE_MSG = "{}: complete future";
+
/**
* Tracks items added to this controller via one of the <i>add</i> methods.
*/
private final FutureManager futures = new FutureManager();
- /**
- * Cancels and stops the pipeline, in that order.
- */
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
- try {
- logger.trace("{}: cancel future", ident(this));
- return super.cancel(mayInterruptIfRunning);
+ return doAndStop(() -> super.cancel(mayInterruptIfRunning), CANCEL_MSG, ident(this));
+ }
- } finally {
- futures.stop();
- }
+ @Override
+ public boolean complete(T value) {
+ return doAndStop(() -> super.complete(value), COMPLETE_MSG, ident(this));
+ }
+
+ @Override
+ public boolean completeExceptionally(Throwable ex) {
+ return doAndStop(() -> super.completeExceptionally(ex), COMPLETE_EXCEPT_MSG, ident(this));
+ }
+
+ @Override
+ public CompletableFuture<T> completeAsync(Supplier<? extends T> supplier, Executor executor) {
+ return super.completeAsync(() -> doAndStop(supplier, COMPLETE_MSG, ident(this)), executor);
+ }
+
+ @Override
+ public CompletableFuture<T> completeAsync(Supplier<? extends T> supplier) {
+ return super.completeAsync(() -> doAndStop(supplier, COMPLETE_MSG, ident(this)));
+ }
+
+ @Override
+ public CompletableFuture<T> completeOnTimeout(T value, long timeout, TimeUnit unit) {
+ logger.info("{}: set future timeout to {} {}", ident(this), timeout, unit);
+ return super.completeOnTimeout(value, timeout, unit);
+ }
+
+ @Override
+ public <U> PipelineControllerFuture<U> newIncompleteFuture() {
+ return new PipelineControllerFuture<>();
}
/**
@@ -67,11 +96,8 @@ public class PipelineControllerFuture<T> extends CompletableFuture<T> {
*
* @return a function that removes the given future
*/
- public <F> BiConsumer<T, Throwable> delayedRemove(Future<F> future) {
- return (value, thrown) -> {
- logger.trace("{}: remove future {}", ident(this), ident(future));
- remove(future);
- };
+ public <F> BiConsumer<F, Throwable> delayedRemove(Future<F> future) {
+ return (value, thrown) -> remove(future);
}
/**
@@ -81,11 +107,8 @@ public class PipelineControllerFuture<T> extends CompletableFuture<T> {
*
* @return a function that removes the given listener
*/
- public BiConsumer<T, Throwable> delayedRemove(Runnable listener) {
- return (value, thrown) -> {
- logger.trace("{}: remove listener {}", ident(this), ident(listener));
- remove(listener);
- };
+ public <F> BiConsumer<F, Throwable> delayedRemove(Runnable listener) {
+ return (value, thrown) -> remove(listener);
}
/**
@@ -98,25 +121,43 @@ public class PipelineControllerFuture<T> extends CompletableFuture<T> {
public BiConsumer<T, Throwable> delayedComplete() {
return (value, thrown) -> {
if (thrown == null) {
- logger.trace("{}: complete and stop future", ident(this));
complete(value);
} else {
- logger.trace("{}: complete exceptionally and stop future", ident(this));
completeExceptionally(thrown);
}
-
- futures.stop();
};
}
/**
+ * Adds a future to the controller and arranges for it to be removed from the
+ * controller when it completes, whether or not it throws an exception. If the
+ * controller has already been stopped, then the future is canceled and a new,
+ * incomplete future is returned.
+ *
+ * @param future future to be wrapped
+ * @return a new future
+ */
+ public CompletableFuture<T> wrap(CompletableFuture<T> future) {
+ if (!isRunning()) {
+ logger.trace("{}: not running, skipping next task {}", ident(this), ident(future));
+ future.cancel(false);
+ return new CompletableFuture<>();
+ }
+
+ add(future);
+ return future.whenComplete(this.delayedRemove(future));
+ }
+
+ /**
* Adds a function whose return value is to be canceled when this controller is
* stopped. Note: if the controller is already stopped, then the function will
* <i>not</i> be executed.
*
- * @param futureMaker function to be invoked in the future
+ * @param futureMaker function to be invoked to create the future
+ * @return a function to create the future and arrange for it to be managed by this
+ * controller
*/
- public <F> Function<F, CompletableFuture<F>> add(Function<F, CompletableFuture<F>> futureMaker) {
+ public <F> Function<F, CompletableFuture<F>> wrap(Function<F, CompletableFuture<F>> futureMaker) {
return input -> {
if (!isRunning()) {
@@ -127,7 +168,7 @@ public class PipelineControllerFuture<T> extends CompletableFuture<T> {
CompletableFuture<F> future = futureMaker.apply(input);
add(future);
- return future;
+ return future.whenComplete(delayedRemove(future));
};
}
@@ -154,4 +195,26 @@ public class PipelineControllerFuture<T> extends CompletableFuture<T> {
logger.trace("{}: remove listener {}", ident(this), ident(listener));
futures.remove(listener);
}
+
+ /**
+ * Performs an operation, stops the futures, and returns the value from the operation.
+ * Logs a message using the given arguments.
+ *
+ *
+ * @param <R> type of value to be returned
+ * @param supplier operation to perform
+ * @param message message to be logged
+ * @param args message arguments to fill "{}" place-holders
+ * @return the operation's result
+ */
+ private <R> R doAndStop(Supplier<R> supplier, String message, Object... args) {
+ try {
+ logger.trace(message, args);
+ return supplier.get();
+
+ } finally {
+ logger.trace("{}: stopping this future", ident(this));
+ futures.stop();
+ }
+ }
}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/AsyncResponseHandlerTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/AsyncResponseHandlerTest.java
new file mode 100644
index 000000000..31c6d2077
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/AsyncResponseHandlerTest.java
@@ -0,0 +1,172 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2020 AT&T Intellectual Property. 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actorserviceprovider;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.policy.controlloop.VirtualControlLoopEvent;
+import org.onap.policy.controlloop.actorserviceprovider.controlloop.ControlLoopEventContext;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
+import org.onap.policy.controlloop.policy.PolicyResult;
+
+public class AsyncResponseHandlerTest {
+
+ private static final String ACTOR = "my-actor";
+ private static final String OPERATION = "my-operation";
+ private static final UUID REQ_ID = UUID.randomUUID();
+ private static final String TEXT = "some text";
+
+ private VirtualControlLoopEvent event;
+ private ControlLoopEventContext context;
+ private ControlLoopOperationParams params;
+ private OperationOutcome outcome;
+ private MyHandler handler;
+
+ /**
+ * Initializes all fields, including {@link #handler}.
+ */
+ @Before
+ public void setUp() {
+ event = new VirtualControlLoopEvent();
+ event.setRequestId(REQ_ID);
+
+ context = new ControlLoopEventContext(event);
+ params = ControlLoopOperationParams.builder().actor(ACTOR).operation(OPERATION).context(context).build();
+ outcome = params.makeOutcome();
+
+ handler = new MyHandler(params, outcome);
+ }
+
+ @Test
+ public void testAsyncResponseHandler_testGetParams_testGetOutcome() {
+ assertSame(params, handler.getParams());
+ assertSame(outcome, handler.getOutcome());
+ }
+
+ @Test
+ public void testHandle() {
+ CompletableFuture<String> future = new CompletableFuture<>();
+ handler.handle(future).complete(outcome);
+
+ assertTrue(future.isCancelled());
+ }
+
+ @Test
+ public void testCompleted() throws Exception {
+ CompletableFuture<OperationOutcome> result = handler.handle(new CompletableFuture<>());
+ handler.completed(TEXT);
+ assertTrue(result.isDone());
+ assertSame(outcome, result.get());
+ assertEquals(PolicyResult.FAILURE_RETRIES, outcome.getResult());
+ assertEquals(TEXT, outcome.getMessage());
+ }
+
+ /**
+ * Tests completed() when doCompleted() throws an exception.
+ */
+ @Test
+ public void testCompletedException() throws Exception {
+ IllegalStateException except = new IllegalStateException();
+
+ outcome = params.makeOutcome();
+ handler = new MyHandler(params, outcome) {
+ @Override
+ protected OperationOutcome doComplete(String rawResponse) {
+ throw except;
+ }
+ };
+
+ CompletableFuture<OperationOutcome> result = handler.handle(new CompletableFuture<>());
+ handler.completed(TEXT);
+ assertTrue(result.isCompletedExceptionally());
+
+ AtomicReference<Throwable> thrown = new AtomicReference<>();
+ result.whenComplete((unused, thrown2) -> thrown.set(thrown2));
+
+ assertSame(except, thrown.get());
+ }
+
+ @Test
+ public void testFailed() throws Exception {
+ IllegalStateException except = new IllegalStateException();
+
+ CompletableFuture<OperationOutcome> result = handler.handle(new CompletableFuture<>());
+ handler.failed(except);
+
+ assertTrue(result.isDone());
+ assertSame(outcome, result.get());
+ assertEquals(PolicyResult.FAILURE_GUARD, outcome.getResult());
+ }
+
+ /**
+ * Tests failed() when doFailed() throws an exception.
+ */
+ @Test
+ public void testFailedException() throws Exception {
+ IllegalStateException except = new IllegalStateException();
+
+ outcome = params.makeOutcome();
+ handler = new MyHandler(params, outcome) {
+ @Override
+ protected OperationOutcome doFailed(Throwable thrown) {
+ throw except;
+ }
+ };
+
+ CompletableFuture<OperationOutcome> result = handler.handle(new CompletableFuture<>());
+ handler.failed(except);
+ assertTrue(result.isCompletedExceptionally());
+
+ AtomicReference<Throwable> thrown = new AtomicReference<>();
+ result.whenComplete((unused, thrown2) -> thrown.set(thrown2));
+
+ assertSame(except, thrown.get());
+ }
+
+ private class MyHandler extends AsyncResponseHandler<String> {
+
+ public MyHandler(ControlLoopOperationParams params, OperationOutcome outcome) {
+ super(params, outcome);
+ }
+
+ @Override
+ protected OperationOutcome doComplete(String rawResponse) {
+ OperationOutcome outcome = getOutcome();
+ outcome.setResult(PolicyResult.FAILURE_RETRIES);
+ outcome.setMessage(rawResponse);
+ return outcome;
+ }
+
+ @Override
+ protected OperationOutcome doFailed(Throwable thrown) {
+ OperationOutcome outcome = getOutcome();
+ outcome.setResult(PolicyResult.FAILURE_GUARD);
+ return outcome;
+ }
+ }
+}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/CallbackManagerTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/CallbackManagerTest.java
new file mode 100644
index 000000000..44606cb14
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/CallbackManagerTest.java
@@ -0,0 +1,89 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2020 AT&T Intellectual Property. 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actorserviceprovider;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.time.Instant;
+import org.junit.Before;
+import org.junit.Test;
+
+public class CallbackManagerTest {
+
+ private CallbackManager mgr;
+
+ @Before
+ public void setUp() {
+ mgr = new CallbackManager();
+ }
+
+ @Test
+ public void testCanStart_testGetStartTime() {
+ // null until canXxx() is called
+ assertNull(mgr.getStartTime());
+
+ assertTrue(mgr.canStart());
+
+ Instant time = mgr.getStartTime();
+ assertNotNull(time);
+ assertNull(mgr.getEndTime());
+
+ // false for now on
+ assertFalse(mgr.canStart());
+ assertFalse(mgr.canStart());
+
+ assertEquals(time, mgr.getStartTime());
+ }
+
+ @Test
+ public void testCanEnd_testGetEndTime() {
+ // null until canXxx() is called
+ assertNull(mgr.getEndTime());
+ assertNull(mgr.getEndTime());
+
+ assertTrue(mgr.canEnd());
+
+ Instant time = mgr.getEndTime();
+ assertNotNull(time);
+ assertNull(mgr.getStartTime());
+
+ // false for now on
+ assertFalse(mgr.canEnd());
+ assertFalse(mgr.canEnd());
+
+ assertEquals(time, mgr.getEndTime());
+ }
+
+ @Test
+ public void testRun() {
+ mgr.run();
+
+ assertNotNull(mgr.getStartTime());
+ assertNotNull(mgr.getEndTime());
+
+ assertFalse(mgr.canStart());
+ assertFalse(mgr.canEnd());
+ }
+}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/OperationOutcomeTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/OperationOutcomeTest.java
new file mode 100644
index 000000000..4e9728336
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/OperationOutcomeTest.java
@@ -0,0 +1,137 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2020 AT&T Intellectual Property. 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actorserviceprovider;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.time.Instant;
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.policy.controlloop.ControlLoopOperation;
+import org.onap.policy.controlloop.policy.PolicyResult;
+
+public class OperationOutcomeTest {
+ private static final String ACTOR = "my-actor";
+ private static final String OPERATION = "my-operation";
+ private static final String TARGET = "my-target";
+ private static final Instant START = Instant.ofEpochMilli(10);
+ private static final Instant END = Instant.ofEpochMilli(20);
+ private static final String SUB_REQ_ID = "my-sub-request-id";
+ private static final PolicyResult RESULT = PolicyResult.FAILURE_GUARD;
+ private static final String MESSAGE = "my-message";
+
+ private OperationOutcome outcome;
+
+ @Before
+ public void setUp() {
+ outcome = new OperationOutcome();
+ }
+
+ @Test
+ public void testOperationOutcomeOperationOutcome() {
+ setAll();
+
+ OperationOutcome outcome2 = new OperationOutcome(outcome);
+
+ assertEquals(ACTOR, outcome2.getActor());
+ assertEquals(OPERATION, outcome2.getOperation());
+ assertEquals(TARGET, outcome2.getTarget());
+ assertEquals(START, outcome2.getStart());
+ assertEquals(END, outcome2.getEnd());
+ assertEquals(SUB_REQ_ID, outcome2.getSubRequestId());
+ assertEquals(RESULT, outcome2.getResult());
+ assertEquals(MESSAGE, outcome2.getMessage());
+ }
+
+ @Test
+ public void testToControlLoopOperation() {
+ setAll();
+
+ ControlLoopOperation outcome2 = outcome.toControlLoopOperation();
+
+ assertEquals(ACTOR, outcome2.getActor());
+ assertEquals(OPERATION, outcome2.getOperation());
+ assertEquals(TARGET, outcome2.getTarget());
+ assertEquals(START, outcome2.getStart());
+ assertEquals(END, outcome2.getEnd());
+ assertEquals(SUB_REQ_ID, outcome2.getSubRequestId());
+ assertEquals(RESULT.toString(), outcome2.getOutcome());
+ assertEquals(MESSAGE, outcome2.getMessage());
+ }
+
+ /**
+ * Tests both isFor() methods, as one invokes the other.
+ */
+ @Test
+ public void testIsFor() {
+ setAll();
+
+ // null case
+ assertFalse(OperationOutcome.isFor(null, ACTOR, OPERATION));
+
+ // actor mismatch
+ assertFalse(OperationOutcome.isFor(outcome, TARGET, OPERATION));
+
+ // operation mismatch
+ assertFalse(OperationOutcome.isFor(outcome, ACTOR, TARGET));
+
+ // null actor in outcome
+ outcome.setActor(null);
+ assertFalse(OperationOutcome.isFor(outcome, ACTOR, OPERATION));
+ outcome.setActor(ACTOR);
+
+ // null operation in outcome
+ outcome.setOperation(null);
+ assertFalse(OperationOutcome.isFor(outcome, ACTOR, OPERATION));
+ outcome.setOperation(OPERATION);
+
+ // null actor argument
+ assertThatThrownBy(() -> outcome.isFor(null, OPERATION));
+
+ // null operation argument
+ assertThatThrownBy(() -> outcome.isFor(ACTOR, null));
+
+ // true case
+ assertTrue(OperationOutcome.isFor(outcome, ACTOR, OPERATION));
+ }
+
+ @Test
+ public void testSetResult() {
+ outcome.setResult(PolicyResult.FAILURE_EXCEPTION);
+ assertEquals(PolicyResult.FAILURE_EXCEPTION, outcome.getResult());
+
+ assertThatThrownBy(() -> outcome.setResult(null));
+ }
+
+ private void setAll() {
+ outcome.setActor(ACTOR);
+ outcome.setEnd(END);
+ outcome.setMessage(MESSAGE);
+ outcome.setOperation(OPERATION);
+ outcome.setResult(RESULT);
+ outcome.setStart(START);
+ outcome.setSubRequestId(SUB_REQ_ID);
+ outcome.setTarget(TARGET);
+ }
+}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/UtilTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/UtilTest.java
index c652e8374..4a3f321cf 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/UtilTest.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/UtilTest.java
@@ -27,15 +27,56 @@ import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
+import ch.qos.logback.classic.Logger;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.Builder;
import lombok.Data;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
import org.junit.Test;
+import org.onap.policy.common.utils.coder.CoderException;
+import org.onap.policy.common.utils.coder.StandardCoder;
+import org.onap.policy.common.utils.test.log.logback.ExtractAppender;
+import org.slf4j.LoggerFactory;
public class UtilTest {
+ private static final String MY_REQUEST = "my-request";
+ private static final String URL = "my-url";
+ private static final String OUT_URL = "OUT|REST|my-url";
+ private static final String IN_URL = "IN|REST|my-url";
+ protected static final String EXPECTED_EXCEPTION = "expected exception";
+
+ /**
+ * Used to attach an appender to the class' logger.
+ */
+ private static final Logger logger = (Logger) LoggerFactory.getLogger(Util.class);
+ private static final ExtractAppender appender = new ExtractAppender();
+
+ /**
+ * Initializes statics.
+ */
+ @BeforeClass
+ public static void setUpBeforeClass() {
+ appender.setContext(logger.getLoggerContext());
+ appender.start();
+
+ logger.addAppender(appender);
+ }
+
+ @AfterClass
+ public static void tearDownAfterClass() {
+ appender.stop();
+ }
+
+ @Before
+ public void setUp() {
+ appender.clearExtractions();
+ }
@Test
public void testIdent() {
@@ -48,11 +89,88 @@ public class UtilTest {
}
@Test
- public void testLogException() {
+ public void testLogRestRequest() throws CoderException {
+ // log structured data
+ appender.clearExtractions();
+ Util.logRestRequest(URL, new Abc(10, null, null));
+ List<String> output = appender.getExtracted();
+ assertEquals(1, output.size());
+
+ assertThat(output.get(0)).contains(OUT_URL).contains("{\n \"intValue\": 10\n}");
+
+ // log a plain string
+ appender.clearExtractions();
+ Util.logRestRequest(URL, MY_REQUEST);
+ output = appender.getExtracted();
+ assertEquals(1, output.size());
+
+ assertThat(output.get(0)).contains(OUT_URL).contains(MY_REQUEST);
+
+ // exception from coder
+ StandardCoder coder = new StandardCoder() {
+ @Override
+ public String encode(Object object, boolean pretty) throws CoderException {
+ throw new CoderException(EXPECTED_EXCEPTION);
+ }
+ };
+
+ appender.clearExtractions();
+ Util.logRestRequest(coder, URL, new Abc(11, null, null));
+ output = appender.getExtracted();
+ assertEquals(2, output.size());
+ assertThat(output.get(0)).contains("cannot pretty-print request");
+ assertThat(output.get(1)).contains(OUT_URL);
+ }
+
+ @Test
+ public void testLogRestResponse() throws CoderException {
+ // log structured data
+ appender.clearExtractions();
+ Util.logRestResponse(URL, new Abc(10, null, null));
+ List<String> output = appender.getExtracted();
+ assertEquals(1, output.size());
+
+ assertThat(output.get(0)).contains(IN_URL).contains("{\n \"intValue\": 10\n}");
+
+ // log null response
+ appender.clearExtractions();
+ Util.logRestResponse(URL, null);
+ output = appender.getExtracted();
+ assertEquals(1, output.size());
+
+ assertThat(output.get(0)).contains(IN_URL).contains("null");
+
+ // log a plain string
+ appender.clearExtractions();
+ Util.logRestResponse(URL, MY_REQUEST);
+ output = appender.getExtracted();
+ assertEquals(1, output.size());
+
+ assertThat(output.get(0)).contains(IN_URL).contains(MY_REQUEST);
+
+ // exception from coder
+ StandardCoder coder = new StandardCoder() {
+ @Override
+ public String encode(Object object, boolean pretty) throws CoderException {
+ throw new CoderException(EXPECTED_EXCEPTION);
+ }
+ };
+
+ appender.clearExtractions();
+ Util.logRestResponse(coder, URL, new Abc(11, null, null));
+ output = appender.getExtracted();
+ assertEquals(2, output.size());
+ assertThat(output.get(0)).contains("cannot pretty-print response");
+ assertThat(output.get(1)).contains(IN_URL);
+ }
+
+ @Test
+ public void testRunFunction() {
// no exception, no log
AtomicInteger count = new AtomicInteger();
- Util.logException(() -> count.incrementAndGet(), "no error");
+ Util.runFunction(() -> count.incrementAndGet(), "no error");
assertEquals(1, count.get());
+ assertEquals(0, appender.getExtracted().size());
// with an exception
Runnable runnable = () -> {
@@ -60,8 +178,17 @@ public class UtilTest {
throw new IllegalStateException("expected exception");
};
- Util.logException(runnable, "error with no args");
- Util.logException(runnable, "error {} {} arg(s)", "with", 1);
+ appender.clearExtractions();
+ Util.runFunction(runnable, "error with no args");
+ List<String> output = appender.getExtracted();
+ assertEquals(1, output.size());
+ assertThat(output.get(0)).contains("error with no args");
+
+ appender.clearExtractions();
+ Util.runFunction(runnable, "error {} {} arg(s)", "with", 2);
+ output = appender.getExtracted();
+ assertEquals(1, output.size());
+ assertThat(output.get(0)).contains("error with 2 arg(s)");
}
@Test
@@ -123,4 +250,14 @@ public class UtilTest {
private int intValue;
private String strValue;
}
+
+ // throws an exception when getXxx() is used
+ public static class DataWithException {
+ @SuppressWarnings("unused")
+ private int intValue;
+
+ public int getIntValue() {
+ throw new IllegalStateException();
+ }
+ }
}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/controlloop/ControlLoopEventContextTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/controlloop/ControlLoopEventContextTest.java
index fcc3fb12e..0d917ad3e 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/controlloop/ControlLoopEventContextTest.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/controlloop/ControlLoopEventContextTest.java
@@ -20,28 +20,55 @@
package org.onap.policy.controlloop.actorserviceprovider.controlloop;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
+import java.util.Map;
+import java.util.UUID;
import org.junit.Before;
import org.junit.Test;
import org.onap.policy.controlloop.VirtualControlLoopEvent;
public class ControlLoopEventContextTest {
+ private static final UUID REQ_ID = UUID.randomUUID();
+ private Map<String, String> enrichment;
private VirtualControlLoopEvent event;
private ControlLoopEventContext context;
+ /**
+ * Initializes data, including {@link #context}.
+ */
@Before
public void setUp() {
+ enrichment = Map.of("abc", "one", "def", "two");
+
event = new VirtualControlLoopEvent();
+ event.setRequestId(REQ_ID);
+ event.setAai(enrichment);
+
context = new ControlLoopEventContext(event);
}
@Test
public void testControlLoopEventContext() {
assertSame(event, context.getEvent());
+ assertSame(REQ_ID, context.getRequestId());
+ assertEquals(enrichment, context.getEnrichment());
+
+ // null event
+ assertThatThrownBy(() -> new ControlLoopEventContext(null));
+
+ // no request id, no enrichment data
+ event.setRequestId(null);
+ event.setAai(null);
+ context = new ControlLoopEventContext(event);
+ assertSame(event, context.getEvent());
+ assertNotNull(context.getRequestId());
+ assertEquals(Map.of(), context.getEnrichment());
}
@Test
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/ActorImplTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/ActorImplTest.java
index 7e0c35a3f..a209fb0d8 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/ActorImplTest.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/ActorImplTest.java
@@ -34,7 +34,6 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-import java.util.Collections;
import java.util.Iterator;
import java.util.Map;
import java.util.TreeMap;
@@ -192,10 +191,10 @@ public class ActorImplTest {
}
@Test
- public void testSetOperators() {
- // cannot set operators if already configured
+ public void testAddOperator() {
+ // cannot add operators if already configured
actor.configure(params);
- assertThatIllegalStateException().isThrownBy(() -> actor.setOperators(Collections.emptyList()));
+ assertThatIllegalStateException().isThrownBy(() -> actor.addOperator(oper1));
/*
* make an actor where operators two and four have names that are duplicates of
@@ -367,7 +366,13 @@ public class ActorImplTest {
* @return a new actor
*/
private ActorImpl makeActor(Operator... operators) {
- return new ActorImpl(ACTOR_NAME, operators);
+ ActorImpl actor = new ActorImpl(ACTOR_NAME);
+
+ for (Operator oper : operators) {
+ actor.addOperator(oper);
+ }
+
+ return actor;
}
private static class MyOper extends OperatorPartial implements Operator {
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpActorTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpActorTest.java
new file mode 100644
index 000000000..2da789989
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpActorTest.java
@@ -0,0 +1,81 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2020 AT&T Intellectual Property. 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actorserviceprovider.impl;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.function.Function;
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.policy.controlloop.actorserviceprovider.Util;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.HttpActorParams;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ParameterValidationRuntimeException;
+
+public class HttpActorTest {
+
+ private static final String ACTOR = "my-actor";
+ private static final String UNKNOWN = "unknown";
+ private static final String CLIENT = "my-client";
+ private static final long TIMEOUT = 10L;
+
+ private HttpActor actor;
+
+ @Before
+ public void setUp() {
+ actor = new HttpActor(ACTOR);
+ }
+
+ @Test
+ public void testMakeOperatorParameters() {
+ HttpActorParams params = new HttpActorParams();
+ params.setClientName(CLIENT);
+ params.setTimeoutSec(TIMEOUT);
+ params.setPath(Map.of("operA", "urlA", "operB", "urlB"));
+
+ final HttpActor prov = new HttpActor(ACTOR);
+ Function<String, Map<String, Object>> maker =
+ prov.makeOperatorParameters(Util.translateToMap(prov.getName(), params));
+
+ assertNull(maker.apply(UNKNOWN));
+
+ // use a TreeMap to ensure the properties are sorted
+ assertEquals("{clientName=my-client, path=urlA, timeoutSec=10}",
+ new TreeMap<>(maker.apply("operA")).toString());
+
+ assertEquals("{clientName=my-client, path=urlB, timeoutSec=10}",
+ new TreeMap<>(maker.apply("operB")).toString());
+
+ // with invalid actor parameters
+ params.setClientName(null);
+ assertThatThrownBy(() -> prov.makeOperatorParameters(Util.translateToMap(prov.getName(), params)))
+ .isInstanceOf(ParameterValidationRuntimeException.class);
+ }
+
+ @Test
+ public void testHttpActor() {
+ assertEquals(ACTOR, actor.getName());
+ assertEquals(ACTOR, actor.getFullName());
+ }
+}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperatorTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperatorTest.java
new file mode 100644
index 000000000..c006cf333
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperatorTest.java
@@ -0,0 +1,104 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2020 AT&T Intellectual Property. 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actorserviceprovider.impl;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Map;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.onap.policy.common.endpoints.http.client.HttpClient;
+import org.onap.policy.common.endpoints.http.client.HttpClientFactory;
+import org.onap.policy.controlloop.actorserviceprovider.Util;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.HttpParams;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ParameterValidationRuntimeException;
+
+public class HttpOperatorTest {
+
+ private static final String ACTOR = "my-actor";
+ private static final String OPERATION = "my-name";
+ private static final String CLIENT = "my-client";
+ private static final String PATH = "my-path";
+ private static final long TIMEOUT = 100;
+
+ @Mock
+ private HttpClient client;
+
+ private HttpOperator oper;
+
+ /**
+ * Initializes fields, including {@link #oper}.
+ */
+ @Before
+ public void setUp() {
+ MockitoAnnotations.initMocks(this);
+
+ oper = new HttpOperator(ACTOR, OPERATION);
+ }
+
+ @Test
+ public void testDoConfigureMapOfStringObject_testGetClient_testGetPath_testGetTimeoutSec() {
+ assertNull(oper.getClient());
+ assertNull(oper.getPath());
+ assertEquals(0L, oper.getTimeoutSec());
+
+ oper = new HttpOperator(ACTOR, OPERATION) {
+ @Override
+ protected HttpClientFactory getClientFactory() {
+ HttpClientFactory factory = mock(HttpClientFactory.class);
+ when(factory.get(CLIENT)).thenReturn(client);
+ return factory;
+ }
+ };
+
+ HttpParams params = HttpParams.builder().clientName(CLIENT).path(PATH).timeoutSec(TIMEOUT).build();
+ Map<String, Object> paramMap = Util.translateToMap(OPERATION, params);
+ oper.configure(paramMap);
+
+ assertSame(client, oper.getClient());
+ assertEquals(PATH, oper.getPath());
+ assertEquals(TIMEOUT, oper.getTimeoutSec());
+
+ // test invalid parameters
+ paramMap.remove("path");
+ assertThatThrownBy(() -> oper.configure(paramMap)).isInstanceOf(ParameterValidationRuntimeException.class);
+ }
+
+ @Test
+ public void testHttpOperator() {
+ assertEquals(ACTOR, oper.getActorName());
+ assertEquals(OPERATION, oper.getName());
+ assertEquals(ACTOR + "." + OPERATION, oper.getFullName());
+ }
+
+ @Test
+ public void testGetClient() {
+ assertNotNull(oper.getClientFactory());
+ }
+}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperatorPartialTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperatorPartialTest.java
index 864ac829a..21bc656f2 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperatorPartialTest.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperatorPartialTest.java
@@ -29,6 +29,7 @@ import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import java.time.Instant;
@@ -36,17 +37,20 @@ import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Queue;
import java.util.TreeMap;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
@@ -58,9 +62,10 @@ import org.junit.Before;
import org.junit.Test;
import org.onap.policy.controlloop.ControlLoopOperation;
import org.onap.policy.controlloop.VirtualControlLoopEvent;
+import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
import org.onap.policy.controlloop.actorserviceprovider.controlloop.ControlLoopEventContext;
import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
-import org.onap.policy.controlloop.policy.Policy;
+import org.onap.policy.controlloop.actorserviceprovider.pipeline.PipelineControllerFuture;
import org.onap.policy.controlloop.policy.PolicyResult;
public class OperatorPartialTest {
@@ -75,14 +80,10 @@ public class OperatorPartialTest {
private static final List<PolicyResult> FAILURE_RESULTS = Arrays.asList(PolicyResult.values()).stream()
.filter(result -> result != PolicyResult.SUCCESS).collect(Collectors.toList());
- private static final List<String> FAILURE_STRINGS =
- FAILURE_RESULTS.stream().map(Object::toString).collect(Collectors.toList());
-
private VirtualControlLoopEvent event;
private Map<String, Object> config;
private ControlLoopEventContext context;
private MyExec executor;
- private Policy policy;
private ControlLoopOperationParams params;
private MyOper oper;
@@ -92,8 +93,8 @@ public class OperatorPartialTest {
private Instant tstart;
- private ControlLoopOperation opstart;
- private ControlLoopOperation opend;
+ private OperationOutcome opstart;
+ private OperationOutcome opend;
/**
* Initializes the fields, including {@link #oper}.
@@ -107,13 +108,9 @@ public class OperatorPartialTest {
context = new ControlLoopEventContext(event);
executor = new MyExec();
- policy = new Policy();
- policy.setActor(ACTOR);
- policy.setRecipe(OPERATOR);
- policy.setTimeout(TIMEOUT);
-
params = ControlLoopOperationParams.builder().completeCallback(this::completer).context(context)
- .executor(executor).policy(policy).startCallback(this::starter).target(TARGET).build();
+ .executor(executor).actor(ACTOR).operation(OPERATOR).timeoutSec(TIMEOUT)
+ .startCallback(this::starter).targetEntity(TARGET).build();
oper = new MyOper();
oper.configure(new TreeMap<>());
@@ -133,6 +130,31 @@ public class OperatorPartialTest {
}
@Test
+ public void testGetBlockingExecutor() throws InterruptedException {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ /*
+ * Use an operator that doesn't override getBlockingExecutor().
+ */
+ OperatorPartial oper2 = new OperatorPartial(ACTOR, OPERATOR) {};
+ oper2.getBlockingExecutor().execute(() -> latch.countDown());
+
+ assertTrue(latch.await(5, TimeUnit.SECONDS));
+ }
+
+ @Test
+ public void testDoConfigure() {
+ oper = spy(new MyOper());
+
+ oper.configure(config);
+ verify(oper).configure(config);
+
+ // repeat - SHOULD be run again
+ oper.configure(config);
+ verify(oper, times(2)).configure(config);
+ }
+
+ @Test
public void testDoStart() {
oper = spy(new MyOper());
@@ -181,7 +203,7 @@ public class OperatorPartialTest {
}
@Test
- public void testStartOperation_testVerifyRunning() {
+ public void testStartOperation() {
verifyRun("testStartOperation", 1, 1, PolicyResult.SUCCESS);
}
@@ -201,17 +223,13 @@ public class OperatorPartialTest {
* Tests startOperation() when the operation has a preprocessor.
*/
@Test
- public void testStartOperationWithPreprocessor_testStartPreprocessor() {
+ public void testStartOperationWithPreprocessor() {
AtomicInteger count = new AtomicInteger();
- // @formatter:off
- Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> preproc =
- oper -> CompletableFuture.supplyAsync(() -> {
- count.incrementAndGet();
- oper.setOutcome(PolicyResult.SUCCESS.toString());
- return oper;
- }, executor);
- // @formatter:on
+ CompletableFuture<OperationOutcome> preproc = CompletableFuture.supplyAsync(() -> {
+ count.incrementAndGet();
+ return makeSuccess();
+ }, executor);
oper.setPreProcessor(preproc);
@@ -233,7 +251,7 @@ public class OperatorPartialTest {
assertNotNull(opstart);
assertNotNull(opend);
- assertEquals(PolicyResult.SUCCESS.toString(), opend.getOutcome());
+ assertEquals(PolicyResult.SUCCESS, opend.getResult());
assertEquals(MAX_PARALLEL_REQUESTS, numStart);
assertEquals(MAX_PARALLEL_REQUESTS, oper.getCount());
@@ -245,11 +263,7 @@ public class OperatorPartialTest {
*/
@Test
public void testStartPreprocessorFailure() {
- // arrange for the preprocessor to return a failure
- oper.setPreProcessor(oper -> {
- oper.setOutcome(PolicyResult.FAILURE_GUARD.toString());
- return CompletableFuture.completedFuture(oper);
- });
+ oper.setPreProcessor(CompletableFuture.completedFuture(makeFailure()));
verifyRun("testStartPreprocessorFailure", 1, 0, PolicyResult.FAILURE_GUARD);
}
@@ -260,9 +274,7 @@ public class OperatorPartialTest {
@Test
public void testStartPreprocessorException() {
// arrange for the preprocessor to throw an exception
- oper.setPreProcessor(oper -> {
- throw new IllegalStateException(EXPECTED_EXCEPTION);
- });
+ oper.setPreProcessor(CompletableFuture.failedFuture(new IllegalStateException(EXPECTED_EXCEPTION)));
verifyRun("testStartPreprocessorException", 1, 0, PolicyResult.FAILURE_GUARD);
}
@@ -273,10 +285,7 @@ public class OperatorPartialTest {
@Test
public void testStartPreprocessorNotRunning() {
// arrange for the preprocessor to return success, which will be ignored
- oper.setPreProcessor(oper -> {
- oper.setOutcome(PolicyResult.SUCCESS.toString());
- return CompletableFuture.completedFuture(oper);
- });
+ oper.setPreProcessor(CompletableFuture.completedFuture(makeSuccess()));
oper.startOperation(params).cancel(false);
assertTrue(executor.runAll());
@@ -296,8 +305,7 @@ public class OperatorPartialTest {
public void testStartPreprocessorBuilderException() {
oper = new MyOper() {
@Override
- protected Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> doPreprocessorAsFuture(
- ControlLoopOperationParams params) {
+ protected CompletableFuture<OperationOutcome> startPreprocessorAsync(ControlLoopOperationParams params) {
throw new IllegalStateException(EXPECTED_EXCEPTION);
}
};
@@ -312,51 +320,27 @@ public class OperatorPartialTest {
}
@Test
- public void testDoPreprocessorAsFuture() {
- assertNull(oper.doPreprocessorAsFuture(params));
+ public void testStartPreprocessorAsync() {
+ assertNull(oper.startPreprocessorAsync(params));
}
@Test
- public void testStartOperationOnly_testDoOperationAsFuture() {
+ public void testStartOperationAsync() {
oper.startOperation(params);
assertTrue(executor.runAll());
assertEquals(1, oper.getCount());
}
- /**
- * Tests startOperationOnce() when
- * {@link OperatorPartial#doOperationAsFuture(ControlLoopOperationParams)} throws an
- * exception.
- */
- @Test
- public void testStartOperationOnceBuilderException() {
- oper = new MyOper() {
- @Override
- protected Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> doOperationAsFuture(
- ControlLoopOperationParams params, int attempt) {
- throw new IllegalStateException(EXPECTED_EXCEPTION);
- }
- };
-
- oper.configure(new TreeMap<>());
- oper.start();
-
- assertThatIllegalStateException().isThrownBy(() -> oper.startOperation(params));
-
- // should be nothing in the queue
- assertEquals(0, executor.getQueueLength());
- }
-
@Test
public void testIsSuccess() {
- ControlLoopOperation outcome = new ControlLoopOperation();
+ OperationOutcome outcome = new OperationOutcome();
- outcome.setOutcome(PolicyResult.SUCCESS.toString());
+ outcome.setResult(PolicyResult.SUCCESS);
assertTrue(oper.isSuccess(outcome));
- for (String failure : FAILURE_STRINGS) {
- outcome.setOutcome(failure);
+ for (PolicyResult failure : FAILURE_RESULTS) {
+ outcome.setResult(failure);
assertFalse("testIsSuccess-" + failure, oper.isSuccess(outcome));
}
}
@@ -365,17 +349,17 @@ public class OperatorPartialTest {
public void testIsActorFailed() {
assertFalse(oper.isActorFailed(null));
- ControlLoopOperation outcome = params.makeOutcome();
+ OperationOutcome outcome = params.makeOutcome();
// incorrect outcome
- outcome.setOutcome(PolicyResult.SUCCESS.toString());
+ outcome.setResult(PolicyResult.SUCCESS);
assertFalse(oper.isActorFailed(outcome));
- outcome.setOutcome(PolicyResult.FAILURE_RETRIES.toString());
+ outcome.setResult(PolicyResult.FAILURE_RETRIES);
assertFalse(oper.isActorFailed(outcome));
// correct outcome
- outcome.setOutcome(PolicyResult.FAILURE.toString());
+ outcome.setResult(PolicyResult.FAILURE);
// incorrect actor
outcome.setActor(TARGET);
@@ -400,7 +384,12 @@ public class OperatorPartialTest {
/*
* Use an operator that doesn't override doOperation().
*/
- OperatorPartial oper2 = new OperatorPartial(ACTOR, OPERATOR) {};
+ OperatorPartial oper2 = new OperatorPartial(ACTOR, OPERATOR) {
+ @Override
+ protected Executor getBlockingExecutor() {
+ return executor;
+ }
+ };
oper2.configure(new TreeMap<>());
oper2.start();
@@ -409,7 +398,7 @@ public class OperatorPartialTest {
assertTrue(executor.runAll());
assertNotNull(opend);
- assertEquals(PolicyResult.FAILURE_EXCEPTION.toString(), opend.getOutcome());
+ assertEquals(PolicyResult.FAILURE_EXCEPTION, opend.getResult());
}
@Test
@@ -421,36 +410,34 @@ public class OperatorPartialTest {
// trigger timeout very quickly
oper = new MyOper() {
@Override
- protected long getTimeOutMillis(Policy policy) {
+ protected long getTimeOutMillis(Integer timeoutSec) {
return 1;
}
@Override
- protected Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> doOperationAsFuture(
- ControlLoopOperationParams params, int attempt) {
-
- return outcome -> {
- ControlLoopOperation outcome2 = params.makeOutcome();
- outcome2.setOutcome(PolicyResult.SUCCESS.toString());
-
- /*
- * Create an incomplete future that will timeout after the operation's
- * timeout. If it fires before the other timer, then it will return a
- * SUCCESS outcome.
- */
- CompletableFuture<ControlLoopOperation> future = new CompletableFuture<>();
- future = future.orTimeout(1, TimeUnit.SECONDS).handleAsync((unused1, unused2) -> outcome,
- params.getExecutor());
-
- return future;
- };
+ protected CompletableFuture<OperationOutcome> startOperationAsync(ControlLoopOperationParams params,
+ int attempt, OperationOutcome outcome) {
+
+ OperationOutcome outcome2 = params.makeOutcome();
+ outcome2.setResult(PolicyResult.SUCCESS);
+
+ /*
+ * Create an incomplete future that will timeout after the operation's
+ * timeout. If it fires before the other timer, then it will return a
+ * SUCCESS outcome.
+ */
+ CompletableFuture<OperationOutcome> future = new CompletableFuture<>();
+ future = future.orTimeout(1, TimeUnit.SECONDS).handleAsync((unused1, unused2) -> outcome,
+ params.getExecutor());
+
+ return future;
}
};
oper.configure(new TreeMap<>());
oper.start();
- assertEquals(PolicyResult.FAILURE_TIMEOUT.toString(), oper.startOperation(params).get().getOutcome());
+ assertEquals(PolicyResult.FAILURE_TIMEOUT, oper.startOperation(params).get().getResult());
}
/**
@@ -466,40 +453,45 @@ public class OperatorPartialTest {
// trigger timeout very quickly
oper = new MyOper() {
@Override
- protected long getTimeOutMillis(Policy policy) {
+ protected long getTimeOutMillis(Integer timeoutSec) {
return 10;
}
@Override
- protected Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> doPreprocessorAsFuture(
- ControlLoopOperationParams params) {
-
- return outcome -> {
- outcome.setOutcome(PolicyResult.SUCCESS.toString());
-
- /*
- * Create an incomplete future that will timeout after the operation's
- * timeout. If it fires before the other timer, then it will return a
- * SUCCESS outcome.
- */
- CompletableFuture<ControlLoopOperation> future = new CompletableFuture<>();
- future = future.orTimeout(200, TimeUnit.MILLISECONDS).handleAsync((unused1, unused2) -> outcome,
- params.getExecutor());
-
- return future;
+ protected Executor getBlockingExecutor() {
+ return command -> {
+ Thread thread = new Thread(command);
+ thread.start();
};
}
+
+ @Override
+ protected CompletableFuture<OperationOutcome> startPreprocessorAsync(ControlLoopOperationParams params) {
+
+ OperationOutcome outcome = makeSuccess();
+
+ /*
+ * Create an incomplete future that will timeout after the operation's
+ * timeout. If it fires before the other timer, then it will return a
+ * SUCCESS outcome.
+ */
+ CompletableFuture<OperationOutcome> future = new CompletableFuture<>();
+ future = future.orTimeout(200, TimeUnit.MILLISECONDS).handleAsync((unused1, unused2) -> outcome,
+ params.getExecutor());
+
+ return future;
+ }
};
oper.configure(new TreeMap<>());
oper.start();
- ControlLoopOperation result = oper.startOperation(params).get();
- assertEquals(PolicyResult.SUCCESS.toString(), result.getOutcome());
+ OperationOutcome result = oper.startOperation(params).get();
+ assertEquals(PolicyResult.SUCCESS, result.getResult());
assertNotNull(opstart);
assertNotNull(opend);
- assertEquals(PolicyResult.SUCCESS.toString(), opend.getOutcome());
+ assertEquals(PolicyResult.SUCCESS, opend.getResult());
assertEquals(1, numStart);
assertEquals(1, oper.getCount());
@@ -510,8 +502,8 @@ public class OperatorPartialTest {
* Tests retry functions, when the count is set to zero and retries are exhausted.
*/
@Test
- public void testSetRetryFlag_testRetryOnFailure_ZeroRetries() {
- policy.setRetry(0);
+ public void testSetRetryFlag_testRetryOnFailure_ZeroRetries_testStartOperationAttempt() {
+ params = params.toBuilder().retry(0).build();
oper.setMaxFailures(10);
verifyRun("testSetRetryFlag_testRetryOnFailure_ZeroRetries", 1, 1, PolicyResult.FAILURE);
@@ -522,7 +514,7 @@ public class OperatorPartialTest {
*/
@Test
public void testSetRetryFlag_testRetryOnFailure_NullRetries() {
- policy.setRetry(null);
+ params = params.toBuilder().retry(null).build();
oper.setMaxFailures(10);
verifyRun("testSetRetryFlag_testRetryOnFailure_NullRetries", 1, 1, PolicyResult.FAILURE);
@@ -534,10 +526,11 @@ public class OperatorPartialTest {
@Test
public void testSetRetryFlag_testRetryOnFailure_RetriesExhausted() {
final int maxRetries = 3;
- policy.setRetry(maxRetries);
+ params = params.toBuilder().retry(maxRetries).build();
oper.setMaxFailures(10);
- verifyRun("testVerifyRunningWhenNot", maxRetries + 1, maxRetries + 1, PolicyResult.FAILURE_RETRIES);
+ verifyRun("testSetRetryFlag_testRetryOnFailure_RetriesExhausted", maxRetries + 1, maxRetries + 1,
+ PolicyResult.FAILURE_RETRIES);
}
/**
@@ -545,7 +538,7 @@ public class OperatorPartialTest {
*/
@Test
public void testSetRetryFlag_testRetryOnFailure_SuccessAfterRetries() {
- policy.setRetry(10);
+ params = params.toBuilder().retry(10).build();
final int maxFailures = 3;
oper.setMaxFailures(maxFailures);
@@ -563,8 +556,8 @@ public class OperatorPartialTest {
// arrange to return null from doOperation()
oper = new MyOper() {
@Override
- protected ControlLoopOperation doOperation(ControlLoopOperationParams params, int attempt,
- ControlLoopOperation operation) {
+ protected OperationOutcome doOperation(ControlLoopOperationParams params, int attempt,
+ OperationOutcome operation) {
// update counters
super.doOperation(params, attempt, operation);
@@ -579,96 +572,55 @@ public class OperatorPartialTest {
}
@Test
- public void testGetActorOutcome() {
- assertNull(oper.getActorOutcome(null));
+ public void testIsSameOperation() {
+ assertFalse(oper.isSameOperation(null));
- ControlLoopOperation outcome = params.makeOutcome();
- outcome.setOutcome(TARGET);
+ OperationOutcome outcome = params.makeOutcome();
- // wrong actor - should be null
+ // wrong actor - should be false
outcome.setActor(null);
- assertNull(oper.getActorOutcome(outcome));
+ assertFalse(oper.isSameOperation(outcome));
outcome.setActor(TARGET);
- assertNull(oper.getActorOutcome(outcome));
+ assertFalse(oper.isSameOperation(outcome));
outcome.setActor(ACTOR);
// wrong operation - should be null
outcome.setOperation(null);
- assertNull(oper.getActorOutcome(outcome));
+ assertFalse(oper.isSameOperation(outcome));
outcome.setOperation(TARGET);
- assertNull(oper.getActorOutcome(outcome));
+ assertFalse(oper.isSameOperation(outcome));
outcome.setOperation(OPERATOR);
- assertEquals(TARGET, oper.getActorOutcome(outcome));
- }
-
- @Test
- public void testOnSuccess() throws Exception {
- AtomicInteger count = new AtomicInteger();
-
- final Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> nextStep = oper -> {
- count.incrementAndGet();
- return CompletableFuture.completedFuture(oper);
- };
-
- // pass it a null outcome
- ControlLoopOperation outcome = oper.onSuccess(params, nextStep).apply(null).get();
- assertNotNull(outcome);
- assertEquals(PolicyResult.FAILURE.toString(), outcome.getOutcome());
- assertEquals(0, count.get());
-
- // pass it an unpopulated (i.e., failed) outcome
- outcome = new ControlLoopOperation();
- assertSame(outcome, oper.onSuccess(params, nextStep).apply(outcome).get());
- assertEquals(0, count.get());
-
- // pass it a successful outcome
- outcome = params.makeOutcome();
- outcome.setOutcome(PolicyResult.SUCCESS.toString());
- assertSame(outcome, oper.onSuccess(params, nextStep).apply(outcome).get());
- assertEquals(PolicyResult.SUCCESS.toString(), outcome.getOutcome());
- assertEquals(1, count.get());
+ assertTrue(oper.isSameOperation(outcome));
}
/**
- * Tests onSuccess() and handleFailure() when the outcome is a success.
+ * Tests handleFailure() when the outcome is a success.
*/
@Test
- public void testOnSuccessTrue_testHandleFailureTrue() {
- // arrange to return a success from the preprocessor
- oper.setPreProcessor(oper -> {
- oper.setOutcome(PolicyResult.SUCCESS.toString());
- return CompletableFuture.completedFuture(oper);
- });
-
- verifyRun("testOnSuccessTrue_testHandleFailureTrue", 1, 1, PolicyResult.SUCCESS);
+ public void testHandlePreprocessorFailureTrue() {
+ oper.setPreProcessor(CompletableFuture.completedFuture(makeSuccess()));
+ verifyRun("testHandlePreprocessorFailureTrue", 1, 1, PolicyResult.SUCCESS);
}
/**
- * Tests onSuccess() and handleFailure() when the outcome is <i>not</i> a success.
+ * Tests handleFailure() when the outcome is <i>not</i> a success.
*/
@Test
- public void testOnSuccessFalse_testHandleFailureFalse() throws Exception {
- // arrange to return a failure from the preprocessor
- oper.setPreProcessor(oper -> {
- oper.setOutcome(PolicyResult.FAILURE.toString());
- return CompletableFuture.completedFuture(oper);
- });
-
- verifyRun("testOnSuccessFalse_testHandleFailureFalse", 1, 0, PolicyResult.FAILURE_GUARD);
+ public void testHandlePreprocessorFailureFalse() throws Exception {
+ oper.setPreProcessor(CompletableFuture.completedFuture(makeFailure()));
+ verifyRun("testHandlePreprocessorFailureFalse", 1, 0, PolicyResult.FAILURE_GUARD);
}
/**
- * Tests onSuccess() and handleFailure() when the outcome is {@code null}.
+ * Tests handleFailure() when the outcome is {@code null}.
*/
@Test
- public void testOnSuccessFalse_testHandleFailureNull() throws Exception {
+ public void testHandlePreprocessorFailureNull() throws Exception {
// arrange to return null from the preprocessor
- oper.setPreProcessor(oper -> {
- return CompletableFuture.completedFuture(null);
- });
+ oper.setPreProcessor(CompletableFuture.completedFuture(null));
- verifyRun("testOnSuccessFalse_testHandleFailureNull", 1, 0, PolicyResult.FAILURE_GUARD);
+ verifyRun("testHandlePreprocessorFailureNull", 1, 0, PolicyResult.FAILURE_GUARD);
}
@Test
@@ -688,11 +640,374 @@ public class OperatorPartialTest {
}
/**
- * Tests verifyRunning() when the pipeline is not running.
+ * Tests both flavors of anyOf(), because one invokes the other.
+ */
+ @Test
+ public void testAnyOf() throws Exception {
+ // first task completes, others do not
+ List<CompletableFuture<OperationOutcome>> tasks = new LinkedList<>();
+
+ final OperationOutcome outcome = params.makeOutcome();
+
+ tasks.add(CompletableFuture.completedFuture(outcome));
+ tasks.add(new CompletableFuture<>());
+ tasks.add(new CompletableFuture<>());
+
+ CompletableFuture<OperationOutcome> result = oper.anyOf(params, tasks);
+ assertTrue(executor.runAll());
+
+ assertTrue(result.isDone());
+ assertSame(outcome, result.get());
+
+ // second task completes, others do not
+ tasks = new LinkedList<>();
+
+ tasks.add(new CompletableFuture<>());
+ tasks.add(CompletableFuture.completedFuture(outcome));
+ tasks.add(new CompletableFuture<>());
+
+ result = oper.anyOf(params, tasks);
+ assertTrue(executor.runAll());
+
+ assertTrue(result.isDone());
+ assertSame(outcome, result.get());
+
+ // third task completes, others do not
+ tasks = new LinkedList<>();
+
+ tasks.add(new CompletableFuture<>());
+ tasks.add(new CompletableFuture<>());
+ tasks.add(CompletableFuture.completedFuture(outcome));
+
+ result = oper.anyOf(params, tasks);
+ assertTrue(executor.runAll());
+
+ assertTrue(result.isDone());
+ assertSame(outcome, result.get());
+ }
+
+ /**
+ * Tests both flavors of allOf(), because one invokes the other.
+ */
+ @Test
+ public void testAllOf() throws Exception {
+ List<CompletableFuture<OperationOutcome>> tasks = new LinkedList<>();
+
+ final OperationOutcome outcome = params.makeOutcome();
+
+ CompletableFuture<OperationOutcome> future1 = new CompletableFuture<>();
+ CompletableFuture<OperationOutcome> future2 = new CompletableFuture<>();
+ CompletableFuture<OperationOutcome> future3 = new CompletableFuture<>();
+
+ tasks.add(future1);
+ tasks.add(future2);
+ tasks.add(future3);
+
+ CompletableFuture<OperationOutcome> result = oper.allOf(params, tasks);
+
+ assertTrue(executor.runAll());
+ assertFalse(result.isDone());
+ future1.complete(outcome);
+
+ // complete 3 before 2
+ assertTrue(executor.runAll());
+ assertFalse(result.isDone());
+ future3.complete(outcome);
+
+ assertTrue(executor.runAll());
+ assertFalse(result.isDone());
+ future2.complete(outcome);
+
+ // all of them are now done
+ assertTrue(executor.runAll());
+ assertTrue(result.isDone());
+ assertSame(outcome, result.get());
+ }
+
+ @Test
+ public void testCombineOutcomes() throws Exception {
+ // only one outcome
+ verifyOutcomes(0, PolicyResult.SUCCESS);
+ verifyOutcomes(0, PolicyResult.FAILURE_EXCEPTION);
+
+ // maximum is in different positions
+ verifyOutcomes(0, PolicyResult.FAILURE, PolicyResult.SUCCESS, PolicyResult.FAILURE_GUARD);
+ verifyOutcomes(1, PolicyResult.SUCCESS, PolicyResult.FAILURE, PolicyResult.FAILURE_GUARD);
+ verifyOutcomes(2, PolicyResult.SUCCESS, PolicyResult.FAILURE_GUARD, PolicyResult.FAILURE);
+
+ // null outcome
+ final List<CompletableFuture<OperationOutcome>> tasks = new LinkedList<>();
+ tasks.add(CompletableFuture.completedFuture(null));
+ CompletableFuture<OperationOutcome> result = oper.allOf(params, tasks);
+
+ assertTrue(executor.runAll());
+ assertTrue(result.isDone());
+ assertNull(result.get());
+
+ // one throws an exception during execution
+ IllegalStateException except = new IllegalStateException(EXPECTED_EXCEPTION);
+
+ tasks.clear();
+ tasks.add(CompletableFuture.completedFuture(params.makeOutcome()));
+ tasks.add(CompletableFuture.failedFuture(except));
+ tasks.add(CompletableFuture.completedFuture(params.makeOutcome()));
+ result = oper.allOf(params, tasks);
+
+ assertTrue(executor.runAll());
+ assertTrue(result.isCompletedExceptionally());
+ result.whenComplete((unused, thrown) -> assertSame(except, thrown));
+ }
+
+ private void verifyOutcomes(int expected, PolicyResult... results) throws Exception {
+ List<CompletableFuture<OperationOutcome>> tasks = new LinkedList<>();
+
+
+ OperationOutcome expectedOutcome = null;
+
+ for (int count = 0; count < results.length; ++count) {
+ OperationOutcome outcome = params.makeOutcome();
+ outcome.setResult(results[count]);
+ tasks.add(CompletableFuture.completedFuture(outcome));
+
+ if (count == expected) {
+ expectedOutcome = outcome;
+ }
+ }
+
+ CompletableFuture<OperationOutcome> result = oper.allOf(params, tasks);
+
+ assertTrue(executor.runAll());
+ assertTrue(result.isDone());
+ assertSame(expectedOutcome, result.get());
+ }
+
+ private Function<OperationOutcome, CompletableFuture<OperationOutcome>> makeTask(
+ final OperationOutcome taskOutcome) {
+
+ return outcome -> CompletableFuture.completedFuture(taskOutcome);
+ }
+
+ @Test
+ public void testDetmPriority() {
+ assertEquals(1, oper.detmPriority(null));
+
+ OperationOutcome outcome = params.makeOutcome();
+
+ Map<PolicyResult, Integer> map = Map.of(PolicyResult.SUCCESS, 0, PolicyResult.FAILURE_GUARD, 2,
+ PolicyResult.FAILURE_RETRIES, 3, PolicyResult.FAILURE, 4, PolicyResult.FAILURE_TIMEOUT, 5,
+ PolicyResult.FAILURE_EXCEPTION, 6);
+
+ for (Entry<PolicyResult, Integer> ent : map.entrySet()) {
+ outcome.setResult(ent.getKey());
+ assertEquals(ent.getKey().toString(), ent.getValue().intValue(), oper.detmPriority(outcome));
+ }
+ }
+
+ /**
+ * Tests doTask(Future) when the controller is not running.
+ */
+ @Test
+ public void testDoTaskFutureNotRunning() throws Exception {
+ CompletableFuture<OperationOutcome> taskFuture = new CompletableFuture<>();
+
+ PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+ controller.complete(params.makeOutcome());
+
+ CompletableFuture<OperationOutcome> future =
+ oper.doTask(params, controller, false, params.makeOutcome(), taskFuture);
+ assertFalse(future.isDone());
+ assertTrue(executor.runAll());
+
+ // should not have run the task
+ assertFalse(future.isDone());
+
+ // should have canceled the task future
+ assertTrue(taskFuture.isCancelled());
+ }
+
+ /**
+ * Tests doTask(Future) when the previous outcome was successful.
+ */
+ @Test
+ public void testDoTaskFutureSuccess() throws Exception {
+ CompletableFuture<OperationOutcome> taskFuture = new CompletableFuture<>();
+ final OperationOutcome taskOutcome = params.makeOutcome();
+
+ PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+
+ CompletableFuture<OperationOutcome> future =
+ oper.doTask(params, controller, true, params.makeOutcome(), taskFuture);
+
+ taskFuture.complete(taskOutcome);
+ assertTrue(executor.runAll());
+
+ assertTrue(future.isDone());
+ assertSame(taskOutcome, future.get());
+
+ // controller should not be done yet
+ assertFalse(controller.isDone());
+ }
+
+ /**
+ * Tests doTask(Future) when the previous outcome was failed.
+ */
+ @Test
+ public void testDoTaskFutureFailure() throws Exception {
+ CompletableFuture<OperationOutcome> taskFuture = new CompletableFuture<>();
+ final OperationOutcome failedOutcome = params.makeOutcome();
+ failedOutcome.setResult(PolicyResult.FAILURE);
+
+ PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+
+ CompletableFuture<OperationOutcome> future = oper.doTask(params, controller, true, failedOutcome, taskFuture);
+ assertFalse(future.isDone());
+ assertTrue(executor.runAll());
+
+ // should not have run the task
+ assertFalse(future.isDone());
+
+ // should have canceled the task future
+ assertTrue(taskFuture.isCancelled());
+
+ // controller SHOULD be done now
+ assertTrue(controller.isDone());
+ assertSame(failedOutcome, controller.get());
+ }
+
+ /**
+ * Tests doTask(Future) when the previous outcome was failed, but not checking
+ * success.
+ */
+ @Test
+ public void testDoTaskFutureUncheckedFailure() throws Exception {
+ CompletableFuture<OperationOutcome> taskFuture = new CompletableFuture<>();
+ final OperationOutcome failedOutcome = params.makeOutcome();
+ failedOutcome.setResult(PolicyResult.FAILURE);
+
+ PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+
+ CompletableFuture<OperationOutcome> future = oper.doTask(params, controller, false, failedOutcome, taskFuture);
+ assertFalse(future.isDone());
+
+ // complete the task
+ OperationOutcome taskOutcome = params.makeOutcome();
+ taskFuture.complete(taskOutcome);
+
+ assertTrue(executor.runAll());
+
+ // should have run the task
+ assertTrue(future.isDone());
+
+ assertTrue(future.isDone());
+ assertSame(taskOutcome, future.get());
+
+ // controller should not be done yet
+ assertFalse(controller.isDone());
+ }
+
+ /**
+ * Tests doTask(Function) when the controller is not running.
+ */
+ @Test
+ public void testDoTaskFunctionNotRunning() throws Exception {
+ AtomicBoolean invoked = new AtomicBoolean();
+
+ Function<OperationOutcome, CompletableFuture<OperationOutcome>> task = outcome -> {
+ invoked.set(true);
+ return CompletableFuture.completedFuture(params.makeOutcome());
+ };
+
+ PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+ controller.complete(params.makeOutcome());
+
+ CompletableFuture<OperationOutcome> future =
+ oper.doTask(params, controller, false, task).apply(params.makeOutcome());
+ assertFalse(future.isDone());
+ assertTrue(executor.runAll());
+
+ // should not have run the task
+ assertFalse(future.isDone());
+
+ // should not have even invoked the task
+ assertFalse(invoked.get());
+ }
+
+ /**
+ * Tests doTask(Function) when the previous outcome was successful.
+ */
+ @Test
+ public void testDoTaskFunctionSuccess() throws Exception {
+ final OperationOutcome taskOutcome = params.makeOutcome();
+
+ final OperationOutcome failedOutcome = params.makeOutcome();
+
+ Function<OperationOutcome, CompletableFuture<OperationOutcome>> task = makeTask(taskOutcome);
+
+ PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+
+ CompletableFuture<OperationOutcome> future = oper.doTask(params, controller, true, task).apply(failedOutcome);
+
+ assertTrue(future.isDone());
+ assertSame(taskOutcome, future.get());
+
+ // controller should not be done yet
+ assertFalse(controller.isDone());
+ }
+
+ /**
+ * Tests doTask(Function) when the previous outcome was failed.
+ */
+ @Test
+ public void testDoTaskFunctionFailure() throws Exception {
+ final OperationOutcome failedOutcome = params.makeOutcome();
+ failedOutcome.setResult(PolicyResult.FAILURE);
+
+ AtomicBoolean invoked = new AtomicBoolean();
+
+ Function<OperationOutcome, CompletableFuture<OperationOutcome>> task = outcome -> {
+ invoked.set(true);
+ return CompletableFuture.completedFuture(params.makeOutcome());
+ };
+
+ PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+
+ CompletableFuture<OperationOutcome> future = oper.doTask(params, controller, true, task).apply(failedOutcome);
+ assertFalse(future.isDone());
+ assertTrue(executor.runAll());
+
+ // should not have run the task
+ assertFalse(future.isDone());
+
+ // should not have even invoked the task
+ assertFalse(invoked.get());
+
+ // controller should have the failed task
+ assertTrue(controller.isDone());
+ assertSame(failedOutcome, controller.get());
+ }
+
+ /**
+ * Tests doTask(Function) when the previous outcome was failed, but not checking
+ * success.
*/
@Test
- public void testVerifyRunningWhenNot() {
- verifyRun("testVerifyRunningWhenNot", 0, 0, PolicyResult.SUCCESS, future -> future.cancel(false));
+ public void testDoTaskFunctionUncheckedFailure() throws Exception {
+ final OperationOutcome taskOutcome = params.makeOutcome();
+
+ final OperationOutcome failedOutcome = params.makeOutcome();
+ failedOutcome.setResult(PolicyResult.FAILURE);
+
+ Function<OperationOutcome, CompletableFuture<OperationOutcome>> task = makeTask(taskOutcome);
+
+ PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+
+ CompletableFuture<OperationOutcome> future = oper.doTask(params, controller, false, task).apply(failedOutcome);
+
+ assertTrue(future.isDone());
+ assertSame(taskOutcome, future.get());
+
+ // controller should not be done yet
+ assertFalse(controller.isDone());
}
/**
@@ -700,7 +1015,7 @@ public class OperatorPartialTest {
*/
@Test
public void testCallbackStartedNotRunning() {
- AtomicReference<Future<ControlLoopOperation>> future = new AtomicReference<>();
+ AtomicReference<Future<OperationOutcome>> future = new AtomicReference<>();
/*
* arrange to stop the controller when the start-callback is invoked, but capture
@@ -723,7 +1038,7 @@ public class OperatorPartialTest {
*/
@Test
public void testCallbackCompletedNotRunning() {
- AtomicReference<Future<ControlLoopOperation>> future = new AtomicReference<>();
+ AtomicReference<Future<OperationOutcome>> future = new AtomicReference<>();
// arrange to stop the controller when the start-callback is invoked
params = params.toBuilder().startCallback(oper -> {
@@ -739,36 +1054,36 @@ public class OperatorPartialTest {
}
@Test
- public void testSetOutcomeControlLoopOperationThrowable() {
+ public void testSetOutcomeControlLoopOperationOutcomeThrowable() {
final CompletionException timex = new CompletionException(new TimeoutException(EXPECTED_EXCEPTION));
- ControlLoopOperation outcome;
+ OperationOutcome outcome;
- outcome = new ControlLoopOperation();
+ outcome = new OperationOutcome();
oper.setOutcome(params, outcome, timex);
assertEquals(ControlLoopOperation.FAILED_MSG, outcome.getMessage());
- assertEquals(PolicyResult.FAILURE_TIMEOUT.toString(), outcome.getOutcome());
+ assertEquals(PolicyResult.FAILURE_TIMEOUT, outcome.getResult());
- outcome = new ControlLoopOperation();
- oper.setOutcome(params, outcome, new IllegalStateException());
+ outcome = new OperationOutcome();
+ oper.setOutcome(params, outcome, new IllegalStateException(EXPECTED_EXCEPTION));
assertEquals(ControlLoopOperation.FAILED_MSG, outcome.getMessage());
- assertEquals(PolicyResult.FAILURE_EXCEPTION.toString(), outcome.getOutcome());
+ assertEquals(PolicyResult.FAILURE_EXCEPTION, outcome.getResult());
}
@Test
- public void testSetOutcomeControlLoopOperationPolicyResult() {
- ControlLoopOperation outcome;
+ public void testSetOutcomeControlLoopOperationOutcomePolicyResult() {
+ OperationOutcome outcome;
- outcome = new ControlLoopOperation();
+ outcome = new OperationOutcome();
oper.setOutcome(params, outcome, PolicyResult.SUCCESS);
assertEquals(ControlLoopOperation.SUCCESS_MSG, outcome.getMessage());
- assertEquals(PolicyResult.SUCCESS.toString(), outcome.getOutcome());
+ assertEquals(PolicyResult.SUCCESS, outcome.getResult());
for (PolicyResult result : FAILURE_RESULTS) {
- outcome = new ControlLoopOperation();
+ outcome = new OperationOutcome();
oper.setOutcome(params, outcome, result);
assertEquals(result.toString(), ControlLoopOperation.FAILED_MSG, outcome.getMessage());
- assertEquals(result.toString(), result.toString(), outcome.getOutcome());
+ assertEquals(result.toString(), result, outcome.getResult());
}
}
@@ -776,7 +1091,7 @@ public class OperatorPartialTest {
public void testIsTimeout() {
final TimeoutException timex = new TimeoutException(EXPECTED_EXCEPTION);
- assertFalse(oper.isTimeout(new IllegalStateException()));
+ assertFalse(oper.isTimeout(new IllegalStateException(EXPECTED_EXCEPTION)));
assertFalse(oper.isTimeout(new IllegalStateException(timex)));
assertFalse(oper.isTimeout(new CompletionException(new IllegalStateException(timex))));
assertFalse(oper.isTimeout(new CompletionException(null)));
@@ -788,19 +1103,19 @@ public class OperatorPartialTest {
@Test
public void testGetTimeOutMillis() {
- assertEquals(TIMEOUT * 1000, oper.getTimeOutMillis(policy));
+ assertEquals(TIMEOUT * 1000, oper.getTimeOutMillis(params.getTimeoutSec()));
- policy.setTimeout(null);
- assertEquals(0, oper.getTimeOutMillis(policy));
+ params = params.toBuilder().timeoutSec(null).build();
+ assertEquals(0, oper.getTimeOutMillis(params.getTimeoutSec()));
}
- private void starter(ControlLoopOperation oper) {
+ private void starter(OperationOutcome oper) {
++numStart;
tstart = oper.getStart();
opstart = oper;
}
- private void completer(ControlLoopOperation oper) {
+ private void completer(OperationOutcome oper) {
++numEnd;
opend = oper;
}
@@ -816,21 +1131,18 @@ public class OperatorPartialTest {
};
}
- /**
- * Verifies a run.
- *
- * @param testName test name
- * @param expectedCallbacks number of callbacks expected
- * @param expectedOperations number of operation invocations expected
- * @param expectedResult expected outcome
- */
- private void verifyRun(String testName, int expectedCallbacks, int expectedOperations,
- PolicyResult expectedResult) {
+ private OperationOutcome makeSuccess() {
+ OperationOutcome outcome = params.makeOutcome();
+ outcome.setResult(PolicyResult.SUCCESS);
- String expectedSubRequestId =
- (expectedResult == PolicyResult.FAILURE_EXCEPTION ? null : String.valueOf(expectedOperations));
+ return outcome;
+ }
- verifyRun(testName, expectedCallbacks, expectedOperations, expectedResult, expectedSubRequestId, noop());
+ private OperationOutcome makeFailure() {
+ OperationOutcome outcome = params.makeOutcome();
+ outcome.setResult(PolicyResult.FAILURE);
+
+ return outcome;
}
/**
@@ -840,17 +1152,14 @@ public class OperatorPartialTest {
* @param expectedCallbacks number of callbacks expected
* @param expectedOperations number of operation invocations expected
* @param expectedResult expected outcome
- * @param manipulator function to modify the future returned by
- * {@link OperatorPartial#startOperation(ControlLoopOperationParams)} before
- * the tasks in the executor are run
*/
- private void verifyRun(String testName, int expectedCallbacks, int expectedOperations, PolicyResult expectedResult,
- Consumer<CompletableFuture<ControlLoopOperation>> manipulator) {
+ private void verifyRun(String testName, int expectedCallbacks, int expectedOperations,
+ PolicyResult expectedResult) {
String expectedSubRequestId =
(expectedResult == PolicyResult.FAILURE_EXCEPTION ? null : String.valueOf(expectedOperations));
- verifyRun(testName, expectedCallbacks, expectedOperations, expectedResult, expectedSubRequestId, manipulator);
+ verifyRun(testName, expectedCallbacks, expectedOperations, expectedResult, expectedSubRequestId, noop());
}
/**
@@ -866,9 +1175,9 @@ public class OperatorPartialTest {
* the tasks in the executor are run
*/
private void verifyRun(String testName, int expectedCallbacks, int expectedOperations, PolicyResult expectedResult,
- String expectedSubRequestId, Consumer<CompletableFuture<ControlLoopOperation>> manipulator) {
+ String expectedSubRequestId, Consumer<CompletableFuture<OperationOutcome>> manipulator) {
- CompletableFuture<ControlLoopOperation> future = oper.startOperation(params);
+ CompletableFuture<OperationOutcome> future = oper.startOperation(params);
manipulator.accept(future);
@@ -880,7 +1189,7 @@ public class OperatorPartialTest {
if (expectedCallbacks > 0) {
assertNotNull(testName, opstart);
assertNotNull(testName, opend);
- assertEquals(testName, expectedResult.toString(), opend.getOutcome());
+ assertEquals(testName, expectedResult, opend.getResult());
assertSame(testName, tstart, opstart.getStart());
assertSame(testName, tstart, opend.getStart());
@@ -901,7 +1210,7 @@ public class OperatorPartialTest {
assertEquals(testName, expectedOperations, oper.getCount());
}
- private static class MyOper extends OperatorPartial {
+ private class MyOper extends OperatorPartial {
@Getter
private int count = 0;
@@ -912,15 +1221,15 @@ public class OperatorPartialTest {
private int maxFailures = 0;
@Setter
- private Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> preProcessor;
+ private CompletableFuture<OperationOutcome> preProcessor;
public MyOper() {
super(ACTOR, OPERATOR);
}
@Override
- protected ControlLoopOperation doOperation(ControlLoopOperationParams params, int attempt,
- ControlLoopOperation operation) {
+ protected OperationOutcome doOperation(ControlLoopOperationParams params, int attempt,
+ OperationOutcome operation) {
++count;
if (genException) {
throw new IllegalStateException(EXPECTED_EXCEPTION);
@@ -929,19 +1238,22 @@ public class OperatorPartialTest {
operation.setSubRequestId(String.valueOf(attempt));
if (count > maxFailures) {
- operation.setOutcome(PolicyResult.SUCCESS.toString());
+ operation.setResult(PolicyResult.SUCCESS);
} else {
- operation.setOutcome(PolicyResult.FAILURE.toString());
+ operation.setResult(PolicyResult.FAILURE);
}
return operation;
}
@Override
- protected Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> doPreprocessorAsFuture(
- ControlLoopOperationParams params) {
+ protected CompletableFuture<OperationOutcome> startPreprocessorAsync(ControlLoopOperationParams params) {
+ return (preProcessor != null ? preProcessor : super.startPreprocessorAsync(params));
+ }
- return (preProcessor != null ? preProcessor : super.doPreprocessorAsFuture(params));
+ @Override
+ protected Executor getBlockingExecutor() {
+ return executor;
}
}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/ControlLoopOperationParamsTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/ControlLoopOperationParamsTest.java
index 0c8e77d38..9dd19d548 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/ControlLoopOperationParamsTest.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/ControlLoopOperationParamsTest.java
@@ -35,6 +35,8 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.util.Map;
+import java.util.TreeMap;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
@@ -47,20 +49,23 @@ import org.junit.Test;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import org.onap.policy.common.parameters.BeanValidationResult;
-import org.onap.policy.controlloop.ControlLoopOperation;
import org.onap.policy.controlloop.VirtualControlLoopEvent;
import org.onap.policy.controlloop.actorserviceprovider.ActorService;
+import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
import org.onap.policy.controlloop.actorserviceprovider.Operator;
import org.onap.policy.controlloop.actorserviceprovider.controlloop.ControlLoopEventContext;
import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams.ControlLoopOperationParamsBuilder;
import org.onap.policy.controlloop.actorserviceprovider.spi.Actor;
-import org.onap.policy.controlloop.policy.Policy;
+import org.onap.policy.controlloop.policy.Target;
public class ControlLoopOperationParamsTest {
private static final String EXPECTED_EXCEPTION = "expected exception";
private static final String ACTOR = "my-actor";
private static final String OPERATION = "my-operation";
- private static final String TARGET = "my-target";
+ private static final Target TARGET = new Target();
+ private static final String TARGET_ENTITY = "my-target";
+ private static final Integer RETRY = 3;
+ private static final Integer TIMEOUT = 100;
private static final UUID REQ_ID = UUID.randomUUID();
@Mock
@@ -70,7 +75,7 @@ public class ControlLoopOperationParamsTest {
private ActorService actorService;
@Mock
- private Consumer<ControlLoopOperation> completer;
+ private Consumer<OperationOutcome> completer;
@Mock
private ControlLoopEventContext context;
@@ -82,19 +87,18 @@ public class ControlLoopOperationParamsTest {
private Executor executor;
@Mock
- private CompletableFuture<ControlLoopOperation> operation;
+ private CompletableFuture<OperationOutcome> operation;
@Mock
private Operator operator;
@Mock
- private Policy policy;
+ private Consumer<OperationOutcome> starter;
- @Mock
- private Consumer<ControlLoopOperation> starter;
+ private Map<String, String> payload;
private ControlLoopOperationParams params;
- private ControlLoopOperation outcome;
+ private OperationOutcome outcome;
/**
@@ -112,12 +116,12 @@ public class ControlLoopOperationParamsTest {
when(context.getEvent()).thenReturn(event);
- when(policy.getActor()).thenReturn(ACTOR);
- when(policy.getRecipe()).thenReturn(OPERATION);
+ payload = new TreeMap<>();
params = ControlLoopOperationParams.builder().actorService(actorService).completeCallback(completer)
- .context(context).executor(executor).policy(policy).startCallback(starter).target(TARGET)
- .build();
+ .context(context).executor(executor).actor(ACTOR).operation(OPERATION).payload(payload)
+ .retry(RETRY).target(TARGET).targetEntity(TARGET_ENTITY).timeoutSec(TIMEOUT)
+ .startCallback(starter).build();
outcome = params.makeOutcome();
}
@@ -130,30 +134,6 @@ public class ControlLoopOperationParamsTest {
}
@Test
- public void testGetActor() {
- assertEquals(ACTOR, params.getActor());
-
- // try with null policy
- assertEquals(ControlLoopOperationParams.UNKNOWN, params.toBuilder().policy(null).build().getActor());
-
- // try with null name in the policy
- when(policy.getActor()).thenReturn(null);
- assertEquals(ControlLoopOperationParams.UNKNOWN, params.getActor());
- }
-
- @Test
- public void testGetOperation() {
- assertEquals(OPERATION, params.getOperation());
-
- // try with null policy
- assertEquals(ControlLoopOperationParams.UNKNOWN, params.toBuilder().policy(null).build().getOperation());
-
- // try with null name in the policy
- when(policy.getRecipe()).thenReturn(null);
- assertEquals(ControlLoopOperationParams.UNKNOWN, params.getOperation());
- }
-
- @Test
public void testGetRequestId() {
assertSame(REQ_ID, params.getRequestId());
@@ -170,20 +150,14 @@ public class ControlLoopOperationParamsTest {
assertEquals(ACTOR, outcome.getActor());
assertEquals(OPERATION, outcome.getOperation());
checkRemainingFields("with actor");
-
- // try again with a null policy
- outcome = params.toBuilder().policy(null).build().makeOutcome();
- assertEquals(ControlLoopOperationParams.UNKNOWN, outcome.getActor());
- assertEquals(ControlLoopOperationParams.UNKNOWN, outcome.getOperation());
- checkRemainingFields("unknown actor");
}
protected void checkRemainingFields(String testName) {
- assertEquals(testName, TARGET, outcome.getTarget());
- assertNotNull(testName, outcome.getStart());
+ assertEquals(testName, TARGET_ENTITY, outcome.getTarget());
+ assertNull(testName, outcome.getStart());
assertNull(testName, outcome.getEnd());
assertNull(testName, outcome.getSubRequestId());
- assertNull(testName, outcome.getOutcome());
+ assertNotNull(testName, outcome.getResult());
assertNull(testName, outcome.getMessage());
}
@@ -239,21 +213,23 @@ public class ControlLoopOperationParamsTest {
@Test
public void testValidateFields() {
+ testValidate("actor", "null", bldr -> bldr.actor(null));
testValidate("actorService", "null", bldr -> bldr.actorService(null));
testValidate("context", "null", bldr -> bldr.context(null));
testValidate("executor", "null", bldr -> bldr.executor(null));
- testValidate("policy", "null", bldr -> bldr.policy(null));
- testValidate("target", "null", bldr -> bldr.target(null));
+ testValidate("operation", "null", bldr -> bldr.operation(null));
+ testValidate("target", "null", bldr -> bldr.targetEntity(null));
// check edge cases
assertTrue(params.toBuilder().build().validate().isValid());
// these can be null
- assertTrue(params.toBuilder().startCallback(null).completeCallback(null).build().validate().isValid());
+ assertTrue(params.toBuilder().payload(null).retry(null).target(null).timeoutSec(null).startCallback(null)
+ .completeCallback(null).build().validate().isValid());
// test with minimal fields
- assertTrue(ControlLoopOperationParams.builder().actorService(actorService).context(context).policy(policy)
- .target(TARGET).build().validate().isValid());
+ assertTrue(ControlLoopOperationParams.builder().actorService(actorService).context(context).actor(ACTOR)
+ .operation(OPERATION).targetEntity(TARGET_ENTITY).build().validate().isValid());
}
private void testValidate(String fieldName, String expected,
@@ -275,7 +251,12 @@ public class ControlLoopOperationParamsTest {
}
@Test
- public void testActorService() {
+ public void testGetActor() {
+ assertSame(ACTOR, params.getActor());
+ }
+
+ @Test
+ public void testGetActorService() {
assertSame(actorService, params.getActorService());
}
@@ -293,8 +274,43 @@ public class ControlLoopOperationParamsTest {
}
@Test
- public void testGetPolicy() {
- assertSame(policy, params.getPolicy());
+ public void testGetOperation() {
+ assertSame(OPERATION, params.getOperation());
+ }
+
+ @Test
+ public void testGetPayload() {
+ assertSame(payload, params.getPayload());
+
+ // should be null when unspecified
+ assertNull(ControlLoopOperationParams.builder().build().getPayload());
+ }
+
+ @Test
+ public void testGetRetry() {
+ assertSame(RETRY, params.getRetry());
+
+ // should be null when unspecified
+ assertNull(ControlLoopOperationParams.builder().build().getRetry());
+ }
+
+ @Test
+ public void testTarget() {
+ assertSame(TARGET, params.getTarget());
+
+ // should be null when unspecified
+ assertNull(ControlLoopOperationParams.builder().build().getTarget());
+ }
+
+ @Test
+ public void testGetTimeoutSec() {
+ assertSame(TIMEOUT, params.getTimeoutSec());
+
+ // should be 300 when unspecified
+ assertEquals(Integer.valueOf(300), ControlLoopOperationParams.builder().build().getTimeoutSec());
+
+ // null should be ok too
+ assertNull(ControlLoopOperationParams.builder().timeoutSec(null).build().getTimeoutSec());
}
@Test
@@ -308,7 +324,7 @@ public class ControlLoopOperationParamsTest {
}
@Test
- public void testGetTarget() {
- assertEquals(TARGET, params.getTarget());
+ public void testGetTargetEntity() {
+ assertEquals(TARGET_ENTITY, params.getTargetEntity());
}
}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpActorParamsTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpActorParamsTest.java
index 1763388f2..6c1f538ec 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpActorParamsTest.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpActorParamsTest.java
@@ -90,6 +90,8 @@ public class HttpActorParamsTest {
@Test
public void testValidate() {
+ assertTrue(params.validate(CONTAINER).isValid());
+
testValidateField("clientName", "null", params2 -> params2.setClientName(null));
testValidateField("path", "null", params2 -> params2.setPath(null));
testValidateField("timeoutSec", "minimum", params2 -> params2.setTimeoutSec(-1));
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpParamsTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpParamsTest.java
index 829c480d1..6cf7328ca 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpParamsTest.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpParamsTest.java
@@ -47,6 +47,8 @@ public class HttpParamsTest {
@Test
public void testValidate() {
+ assertTrue(params.validate(CONTAINER).isValid());
+
testValidateField("clientName", "null", bldr -> bldr.clientName(null));
testValidateField("path", "null", bldr -> bldr.path(null));
testValidateField("timeoutSec", "minimum", bldr -> bldr.timeoutSec(-1));
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/PipelineControllerFutureTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/PipelineControllerFutureTest.java
index b421c1ce2..a6b11ef65 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/PipelineControllerFutureTest.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/PipelineControllerFutureTest.java
@@ -23,21 +23,27 @@ package org.onap.policy.controlloop.actorserviceprovider.pipeline;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import java.util.function.Function;
import org.junit.Before;
import org.junit.Test;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
@@ -58,9 +64,10 @@ public class PipelineControllerFutureTest {
private Future<String> future2;
@Mock
- private CompletableFuture<String> compFuture;
+ private Executor executor;
+ private CompletableFuture<String> compFuture;
private PipelineControllerFuture<String> controller;
@@ -72,6 +79,8 @@ public class PipelineControllerFutureTest {
public void setUp() {
MockitoAnnotations.initMocks(this);
+ compFuture = spy(new CompletableFuture<>());
+
controller = new PipelineControllerFuture<>();
controller.add(runnable1);
@@ -89,10 +98,7 @@ public class PipelineControllerFutureTest {
assertTrue(controller.isCancelled());
assertFalse(controller.isRunning());
- verify(runnable1).run();
- verify(runnable2).run();
- verify(future1).cancel(anyBoolean());
- verify(future2).cancel(anyBoolean());
+ verifyStopped();
// re-invoke; nothing should change
assertTrue(controller.cancel(true));
@@ -100,10 +106,155 @@ public class PipelineControllerFutureTest {
assertTrue(controller.isCancelled());
assertFalse(controller.isRunning());
- verify(runnable1).run();
- verify(runnable2).run();
- verify(future1).cancel(anyBoolean());
- verify(future2).cancel(anyBoolean());
+ verifyStopped();
+ }
+
+ @Test
+ public void testCompleteT() throws Exception {
+ assertTrue(controller.complete(TEXT));
+ assertEquals(TEXT, controller.get());
+
+ verifyStopped();
+
+ // repeat - disallowed
+ assertFalse(controller.complete(TEXT));
+ }
+
+ @Test
+ public void testCompleteExceptionallyThrowable() {
+ assertTrue(controller.completeExceptionally(EXPECTED_EXCEPTION));
+ assertThatThrownBy(() -> controller.get()).hasCause(EXPECTED_EXCEPTION);
+
+ verifyStopped();
+
+ // repeat - disallowed
+ assertFalse(controller.completeExceptionally(EXPECTED_EXCEPTION));
+ }
+
+ @Test
+ public void testCompleteAsyncSupplierOfQextendsTExecutor() throws Exception {
+ CompletableFuture<String> future = controller.completeAsync(() -> TEXT, executor);
+
+ // haven't stopped anything yet
+ assertFalse(future.isDone());
+ verify(runnable1, never()).run();
+
+ // get the operation and run it
+ ArgumentCaptor<Runnable> captor = ArgumentCaptor.forClass(Runnable.class);
+ verify(executor).execute(captor.capture());
+ captor.getValue().run();
+
+ // should be done now
+ assertTrue(future.isDone());
+
+ assertEquals(TEXT, future.get());
+
+ verifyStopped();
+ }
+
+ /**
+ * Tests completeAsync(executor) when canceled before execution.
+ */
+ @Test
+ public void testCompleteAsyncSupplierOfQextendsTExecutorCanceled() throws Exception {
+ CompletableFuture<String> future = controller.completeAsync(() -> TEXT, executor);
+
+ assertTrue(future.cancel(false));
+
+ verifyStopped();
+
+ assertTrue(future.isDone());
+
+ assertThatThrownBy(() -> controller.get()).isInstanceOf(CancellationException.class);
+ }
+
+ @Test
+ public void testCompleteAsyncSupplierOfQextendsT() throws Exception {
+ CompletableFuture<String> future = controller.completeAsync(() -> TEXT);
+ assertEquals(TEXT, future.get());
+
+ verifyStopped();
+ }
+
+ /**
+ * Tests completeAsync() when canceled.
+ */
+ @Test
+ public void testCompleteAsyncSupplierOfQextendsTCanceled() throws Exception {
+ CountDownLatch canceled = new CountDownLatch(1);
+
+ // run async, but await until canceled
+ CompletableFuture<String> future = controller.completeAsync(() -> {
+ try {
+ canceled.await();
+ } catch (InterruptedException e) {
+ // do nothing
+ }
+
+ return TEXT;
+ });
+
+ assertTrue(future.cancel(false));
+
+ // let the future run now
+ canceled.countDown();
+
+ verifyStopped();
+
+ assertTrue(future.isDone());
+
+ assertThatThrownBy(() -> controller.get()).isInstanceOf(CancellationException.class);
+ }
+
+ @Test
+ public void testCompleteOnTimeoutTLongTimeUnit() throws Exception {
+ CountDownLatch stopped = new CountDownLatch(1);
+ controller.add(() -> stopped.countDown());
+
+ CompletableFuture<String> future = controller.completeOnTimeout(TEXT, 1, TimeUnit.MILLISECONDS);
+
+ assertEquals(TEXT, future.get());
+
+ /*
+ * Must use latch instead of verifyStopped(), because the runnables may be
+ * executed asynchronously.
+ */
+ assertTrue(stopped.await(5, TimeUnit.SECONDS));
+ }
+
+ /**
+ * Tests completeOnTimeout() when completed before the timeout.
+ */
+ @Test
+ public void testCompleteOnTimeoutTLongTimeUnitNoTimeout() throws Exception {
+ CompletableFuture<String> future = controller.completeOnTimeout("timed out", 5, TimeUnit.SECONDS);
+ controller.complete(TEXT);
+
+ assertEquals(TEXT, future.get());
+
+ verifyStopped();
+ }
+
+ /**
+ * Tests completeOnTimeout() when canceled before the timeout.
+ */
+ @Test
+ public void testCompleteOnTimeoutTLongTimeUnitCanceled() {
+ CompletableFuture<String> future = controller.completeOnTimeout(TEXT, 5, TimeUnit.SECONDS);
+ assertTrue(future.cancel(true));
+
+ assertThatThrownBy(() -> controller.get()).isInstanceOf(CancellationException.class);
+
+ verifyStopped();
+ }
+
+ @Test
+ public void testNewIncompleteFuture() {
+ PipelineControllerFuture<String> future = controller.newIncompleteFuture();
+ assertNotNull(future);
+ assertTrue(future instanceof PipelineControllerFuture);
+ assertNotSame(controller, future);
+ assertFalse(future.isDone());
}
@Test
@@ -208,22 +359,81 @@ public class PipelineControllerFutureTest {
verify(future2).cancel(anyBoolean());
}
+ /**
+ * Tests both wrap() methods.
+ */
@Test
- public void testAddFunction() {
- AtomicReference<String> value = new AtomicReference<>();
+ public void testWrap() throws Exception {
+ controller = spy(controller);
- Function<String, CompletableFuture<String>> func = controller.add(input -> {
- value.set(input);
+ CompletableFuture<String> future = controller.wrap(compFuture);
+ verify(controller, never()).remove(compFuture);
+
+ compFuture.complete(TEXT);
+ assertEquals(TEXT, future.get());
+
+ verify(controller).remove(compFuture);
+ }
+
+ /**
+ * Tests wrap(), when the controller is not running.
+ */
+ @Test
+ public void testWrapNotRunning() throws Exception {
+ controller.cancel(false);
+ controller = spy(controller);
+
+ assertFalse(controller.wrap(compFuture).isDone());
+ verify(controller, never()).add(compFuture);
+ verify(controller, never()).remove(compFuture);
+
+ verify(compFuture).cancel(anyBoolean());
+ }
+
+ /**
+ * Tests wrap(), when the future throws an exception.
+ */
+ @Test
+ public void testWrapException() throws Exception {
+ controller = spy(controller);
+
+ CompletableFuture<String> future = controller.wrap(compFuture);
+ verify(controller, never()).remove(compFuture);
+
+ compFuture.completeExceptionally(EXPECTED_EXCEPTION);
+ assertThatThrownBy(() -> future.get()).hasCause(EXPECTED_EXCEPTION);
+
+ verify(controller).remove(compFuture);
+ }
+
+ @Test
+ public void testWrapFunction() throws Exception {
+
+ Function<String, CompletableFuture<String>> func = controller.wrap(input -> {
+ compFuture.complete(input);
return compFuture;
});
- assertSame(compFuture, func.apply(TEXT));
- assertEquals(TEXT, value.get());
+ CompletableFuture<String> future = func.apply(TEXT);
+ assertTrue(compFuture.isDone());
- verify(compFuture, never()).cancel(anyBoolean());
+ assertEquals(TEXT, future.get());
// should not have completed the controller
assertFalse(controller.isDone());
+ }
+
+ /**
+ * Tests add(Function) when the controller is canceled after the future is added.
+ */
+ @Test
+ public void testWrapFunctionCancel() throws Exception {
+ Function<String, CompletableFuture<String>> func = controller.wrap(input -> compFuture);
+
+ CompletableFuture<String> future = func.apply(TEXT);
+ assertFalse(future.isDone());
+
+ assertFalse(compFuture.isDone());
// cancel - should propagate
controller.cancel(false);
@@ -235,10 +445,10 @@ public class PipelineControllerFutureTest {
* Tests add(Function) when the controller is not running.
*/
@Test
- public void testAddFunctionNotRunning() {
+ public void testWrapFunctionNotRunning() {
AtomicReference<String> value = new AtomicReference<>();
- Function<String, CompletableFuture<String>> func = controller.add(input -> {
+ Function<String, CompletableFuture<String>> func = controller.wrap(input -> {
value.set(input);
return compFuture;
});
@@ -251,4 +461,11 @@ public class PipelineControllerFutureTest {
assertNull(value.get());
}
+
+ private void verifyStopped() {
+ verify(runnable1).run();
+ verify(runnable2).run();
+ verify(future1).cancel(anyBoolean());
+ verify(future2).cancel(anyBoolean());
+ }
}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/resources/logback-test.xml b/models-interactions/model-actors/actorServiceProvider/src/test/resources/logback-test.xml
index f8a1e5112..c7fe46e47 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/resources/logback-test.xml
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/resources/logback-test.xml
@@ -35,8 +35,8 @@
<appender-ref ref="STDOUT" />
</root>
- <!-- this is just an example -->
- <logger name="ch.qos.logback.classic" level="off" additivity="false">
+ <!-- this is required for UtilTest -->
+ <logger name="org.onap.policy.controlloop.actorserviceprovider.Util" level="info" additivity="false">
<appender-ref ref="STDOUT" />
</logger>
</configuration>