Add more code to facilitate actor implementation
Added obtain() to Context
Added startGuardAsync(), in anticipation of adding guards.
Moved logRestXxx() from Util to HttpOperation.
Added actor.test to facilitate testing of actors.
Changed timeoutSec from long to int in various places.
Made a couple of methods public to support junit testing.
Most of the methods required Params to be passed, which indicated a
design issue. Split Operator and Operation so that the Params could
be kept in a field and thus need not be passed to every method.
Basically, renamed OperatorPartial.java to OperationPartial.java and
created a new OperatorPartial.java. Of course, this makes it look to
gerrit like it's all new code, when in fact, most of it is unchanged,
other than removing the Params argument to the method calls. That
accounts for about half of the "lines changed" count.
Issue-ID: POLICY-1625
Change-Id: I9e98c9dadcbed145bf84deb06c9db1c864a3c24a
Signed-off-by: Jim Hahn <jrh3@att.com>
diff --git a/models-interactions/model-actors/actor.sdnc/pom.xml b/models-interactions/model-actors/actor.sdnc/pom.xml
index 04040db..4bb03ec 100644
--- a/models-interactions/model-actors/actor.sdnc/pom.xml
+++ b/models-interactions/model-actors/actor.sdnc/pom.xml
@@ -19,58 +19,71 @@
============LICENSE_END=========================================================
-->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
- <parent>
- <groupId>org.onap.policy.models.policy-models-interactions.model-actors</groupId>
- <artifactId>model-actors</artifactId>
- <version>2.2.1-SNAPSHOT</version>
- </parent>
+ <parent>
+ <groupId>org.onap.policy.models.policy-models-interactions.model-actors</groupId>
+ <artifactId>model-actors</artifactId>
+ <version>2.2.1-SNAPSHOT</version>
+ </parent>
- <artifactId>actor.sdnc</artifactId>
+ <artifactId>actor.sdnc</artifactId>
- <dependencies>
- <dependency>
- <groupId>org.onap.policy.models.policy-models-interactions.model-actors</groupId>
- <artifactId>actorServiceProvider</artifactId>
- <version>${project.version}</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.onap.policy.models.policy-models-interactions.model-impl</groupId>
- <artifactId>sdnc</artifactId>
- <version>${project.version}</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.onap.policy.models.policy-models-interactions.model-impl</groupId>
- <artifactId>events</artifactId>
- <version>${project.version}</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.onap.policy.models.policy-models-interactions.model-impl</groupId>
- <artifactId>aai</artifactId>
- <version>${project.version}</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.onap.policy.common</groupId>
- <artifactId>policy-endpoints</artifactId>
- <version>${policy.common.version}</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>junit</groupId>
- <artifactId>junit</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.onap.policy.models.policy-models-interactions</groupId>
- <artifactId>simulators</artifactId>
- <version>${project.version}</version>
- <scope>test</scope>
- </dependency>
- </dependencies>
+ <dependencies>
+ <dependency>
+ <groupId>org.onap.policy.models.policy-models-interactions.model-actors</groupId>
+ <artifactId>actorServiceProvider</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.onap.policy.models.policy-models-interactions.model-impl</groupId>
+ <artifactId>sdnc</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.onap.policy.models.policy-models-interactions.model-impl</groupId>
+ <artifactId>events</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.onap.policy.models.policy-models-interactions.model-impl</groupId>
+ <artifactId>aai</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.onap.policy.common</groupId>
+ <artifactId>policy-endpoints</artifactId>
+ <version>${policy.common.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.onap.policy.models.policy-models-interactions.model-actors</groupId>
+ <artifactId>actor.test</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.onap.policy.models.policy-models-interactions</groupId>
+ <artifactId>simulators</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.powermock</groupId>
+ <artifactId>powermock-api-mockito2</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
</project>
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/BandwidthOnDemandOperation.java
similarity index 87%
rename from models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/BandwidthOnDemandOperator.java
rename to models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/BandwidthOnDemandOperation.java
index 2927bd8..26cdfad 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/BandwidthOnDemandOperation.java
@@ -23,6 +23,8 @@
import java.util.UUID;
import org.apache.commons.lang3.StringUtils;
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.sdnc.SdncHealRequest;
import org.onap.policy.sdnc.SdncHealRequestHeaderInfo;
import org.onap.policy.sdnc.SdncHealRequestInfo;
@@ -34,7 +36,7 @@
import org.onap.policy.sdnc.SdncHealVnfInfo;
import org.onap.policy.sdnc.SdncRequest;
-public class BandwidthOnDemandOperator extends SdncOperator {
+public class BandwidthOnDemandOperation extends SdncOperation {
public static final String NAME = "BandwidthOnDemand";
public static final String URI = "/GENERIC-RESOURCE-API:vf-module-topology-operation";
@@ -46,14 +48,17 @@
/**
* Constructs the object.
*
- * @param actorName name of the actor with which this operator is associated
+ * @param params operation parameters
+ * @param operator operator that created this operation
*/
- public BandwidthOnDemandOperator(String actorName) {
- super(actorName, NAME);
+ public BandwidthOnDemandOperation(ControlLoopOperationParams params, HttpOperator operator) {
+ super(params, operator);
}
@Override
- protected SdncRequest constructRequest(ControlLoopEventContext context) {
+ protected SdncRequest makeRequest(int attempt) {
+ ControlLoopEventContext context = params.getContext();
+
String serviceInstance = context.getEnrichment().get(SERVICE_ID_KEY);
if (StringUtils.isBlank(serviceInstance)) {
throw new IllegalArgumentException("missing enrichment data, " + SERVICE_ID_KEY);
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/RerouteOperation.java
similarity index 85%
rename from models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/RerouteOperator.java
rename to models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/RerouteOperation.java
index da400f8..f255f3e 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/RerouteOperation.java
@@ -23,6 +23,8 @@
import java.util.UUID;
import org.apache.commons.lang3.StringUtils;
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.sdnc.SdncHealNetworkInfo;
import org.onap.policy.sdnc.SdncHealRequest;
import org.onap.policy.sdnc.SdncHealRequestHeaderInfo;
@@ -30,7 +32,7 @@
import org.onap.policy.sdnc.SdncHealServiceInfo;
import org.onap.policy.sdnc.SdncRequest;
-public class RerouteOperator extends SdncOperator {
+public class RerouteOperation extends SdncOperation {
public static final String NAME = "Reroute";
public static final String URI = "/GENERIC-RESOURCE-API:network-topology-operation";
@@ -42,14 +44,17 @@
/**
* Constructs the object.
*
- * @param actorName name of the actor with which this operator is associated
+ * @param params operation parameters
+ * @param operator operator that created this operation
*/
- public RerouteOperator(String actorName) {
- super(actorName, NAME);
+ public RerouteOperation(ControlLoopOperationParams params, HttpOperator operator) {
+ super(params, operator);
}
@Override
- protected SdncRequest constructRequest(ControlLoopEventContext context) {
+ protected SdncRequest makeRequest(int attempt) {
+ ControlLoopEventContext context = params.getContext();
+
String serviceInstance = context.getEnrichment().get(SERVICE_ID_KEY);
if (StringUtils.isBlank(serviceInstance)) {
throw new IllegalArgumentException("missing enrichment data, " + SERVICE_ID_KEY);
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 8dc8ba5..99a4fda 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
@@ -30,6 +30,7 @@
import org.onap.policy.controlloop.ControlLoopOperation;
import org.onap.policy.controlloop.VirtualControlLoopEvent;
import org.onap.policy.controlloop.actorserviceprovider.impl.HttpActor;
+import org.onap.policy.controlloop.actorserviceprovider.impl.HttpOperator;
import org.onap.policy.controlloop.policy.Policy;
import org.onap.policy.sdnc.SdncHealNetworkInfo;
import org.onap.policy.sdnc.SdncHealRequest;
@@ -76,8 +77,11 @@
public SdncActorServiceProvider() {
super(NAME);
- addOperator(new RerouteOperator(NAME));
- addOperator(new BandwidthOnDemandOperator(NAME));
+ addOperator(HttpOperator.makeOperator(NAME, RerouteOperation.NAME,
+ RerouteOperation::new));
+
+ addOperator(HttpOperator.makeOperator(NAME, BandwidthOnDemandOperation.NAME,
+ BandwidthOnDemandOperation::new));
}
diff --git a/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/SdncOperation.java b/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/SdncOperation.java
new file mode 100644
index 0000000..9d42c49
--- /dev/null
+++ b/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/SdncOperation.java
@@ -0,0 +1,85 @@
+/*-
+ * ============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 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 org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
+import org.onap.policy.controlloop.actorserviceprovider.impl.HttpOperation;
+import org.onap.policy.controlloop.actorserviceprovider.impl.HttpOperator;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
+import org.onap.policy.sdnc.SdncRequest;
+import org.onap.policy.sdnc.SdncResponse;
+
+/**
+ * Superclass for SDNC Operators.
+ */
+public abstract class SdncOperation extends HttpOperation<SdncResponse> {
+
+ /**
+ * Constructs the object.
+ *
+ * @param params operation parameters
+ * @param operator operator that created this operation
+ */
+ public SdncOperation(ControlLoopOperationParams params, HttpOperator operator) {
+ super(params, operator, SdncResponse.class);
+ }
+
+ @Override
+ protected CompletableFuture<OperationOutcome> startOperationAsync(int attempt, OperationOutcome outcome) {
+
+ SdncRequest request = makeRequest(attempt);
+
+ Entity<SdncRequest> entity = Entity.entity(request, MediaType.APPLICATION_JSON);
+
+ Map<String, Object> headers = makeHeaders();
+
+ headers.put("Accept", MediaType.APPLICATION_JSON);
+ String url = makeUrl();
+
+ logRestRequest(url, request);
+
+ // @formatter:off
+ return handleResponse(outcome, url,
+ callback -> operator.getClient().post(callback, makePath(), entity, headers));
+ // @formatter:on
+ }
+
+ /**
+ * Makes the request.
+ *
+ * @param attempt current attempt, starting with "1"
+ * @return a new request to be posted
+ */
+ protected abstract SdncRequest makeRequest(int attempt);
+
+ /**
+ * Checks that the response has an "output" and that the output indicates success.
+ */
+ @Override
+ protected boolean isSuccess(Response rawResponse, SdncResponse response) {
+ return response.getResponseOutput() != null && "200".equals(response.getResponseOutput().getResponseCode());
+ }
+}
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
deleted file mode 100644
index 479ee90..0000000
--- a/models-interactions/model-actors/actor.sdnc/src/main/java/org/onap/policy/controlloop/actor/sdnc/SdncOperator.java
+++ /dev/null
@@ -1,148 +0,0 @@
-/*-
- * ============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 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 org.onap.policy.common.endpoints.http.client.HttpClient;
-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.controlloop.ControlLoopEventContext;
-import org.onap.policy.controlloop.actorserviceprovider.impl.HttpOperator;
-import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
-import org.onap.policy.controlloop.policy.PolicyResult;
-import org.onap.policy.sdnc.SdncRequest;
-import org.onap.policy.sdnc.SdncResponse;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Superclass for SDNC Operators.
- */
-public abstract class SdncOperator extends HttpOperator {
- private static final Logger logger = LoggerFactory.getLogger(SdncOperator.class);
-
- /**
- * Constructs the object.
- *
- * @param actorName name of the actor with which this operator is associated
- * @param name operation name
- */
- public SdncOperator(String actorName, String name) {
- super(actorName, name);
- }
-
- @Override
- protected CompletableFuture<OperationOutcome> startOperationAsync(ControlLoopOperationParams params, int attempt,
- OperationOutcome outcome) {
-
- SdncRequest request = constructRequest(params.getContext());
- return postRequest(params, outcome, request);
- }
-
- /**
- * Constructs the request.
- *
- * @param context associated event context
- * @return a new request
- */
- protected abstract SdncRequest constructRequest(ControlLoopEventContext context);
-
- /**
- * 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 CompletableFuture<OperationOutcome> postRequest(ControlLoopOperationParams params, OperationOutcome outcome,
- SdncRequest sdncRequest) {
- Map<String, Object> headers = new HashMap<>();
-
- headers.put("Accept", "application/json");
- String sdncUrl = getClient().getBaseUrl();
-
- Util.logRestRequest(sdncUrl, sdncRequest);
-
- Entity<SdncRequest> entity = Entity.entity(sdncRequest, MediaType.APPLICATION_JSON);
-
- 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;
-
- public ResponseHandler(ControlLoopOperationParams params, OperationOutcome outcome, String sdncUrl) {
- super(params, outcome);
- this.sdncUrl = sdncUrl;
- }
-
- /**
- * 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);
- }
- }
-
- /**
- * 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
-
- 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
index 02931a4..0623df2 100644
--- 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
@@ -26,45 +26,51 @@
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 {
+public class BandwidthOnDemandOperatorTest extends BasicSdncOperator {
- private BandwidthOnDemandOperator oper;
+ private BandwidthOnDemandOperation oper;
+ public BandwidthOnDemandOperatorTest() {
+ super(DEFAULT_ACTOR, BandwidthOnDemandOperation.NAME);
+ }
/**
* Set up.
*/
@Before
- public void setUp() {
- makeContext();
- oper = new BandwidthOnDemandOperator(ACTOR);
+ public void setUp() throws Exception {
+ super.setUp();
+ oper = new BandwidthOnDemandOperation(params, operator);
}
@Test
public void testBandwidthOnDemandOperator() {
- assertEquals(ACTOR, oper.getActorName());
- assertEquals(BandwidthOnDemandOperator.NAME, oper.getName());
+ assertEquals(DEFAULT_ACTOR, oper.getActorName());
+ assertEquals(BandwidthOnDemandOperation.NAME, oper.getName());
}
@Test
- public void testConstructRequest() throws CoderException {
- SdncRequest request = oper.constructRequest(context);
+ public void testMakeRequest() throws Exception {
+ SdncRequest request = oper.makeRequest(1);
assertEquals("my-service", request.getNsInstanceId());
assertEquals(REQ_ID, request.getRequestId());
- assertEquals(BandwidthOnDemandOperator.URI, request.getUrl());
+ assertEquals(BandwidthOnDemandOperation.URI, request.getUrl());
assertNotNull(request.getHealRequest().getRequestHeaderInfo().getSvcRequestId());
verifyRequest("bod.json", request);
- verifyMissing(oper, BandwidthOnDemandOperator.SERVICE_ID_KEY, "service");
+ verifyMissing(BandwidthOnDemandOperation.SERVICE_ID_KEY, "service", BandwidthOnDemandOperation::new);
+
+ // perform the operation
+ makeContext();
+ verifyRequest("bod.json", verifyOperation(oper));
}
@Override
protected Map<String, String> makeEnrichment() {
- return Map.of(BandwidthOnDemandOperator.SERVICE_ID_KEY, "my-service", BandwidthOnDemandOperator.VNF_ID,
+ return Map.of(BandwidthOnDemandOperation.SERVICE_ID_KEY, "my-service", BandwidthOnDemandOperation.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
deleted file mode 100644
index b9028d4..0000000
--- a/models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/BasicOperator.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*-
- * ============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/BasicSdncOperator.java b/models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/BasicSdncOperator.java
new file mode 100644
index 0000000..d8c707c
--- /dev/null
+++ b/models-interactions/model-actors/actor.sdnc/src/test/java/org/onap/policy/controlloop/actor/sdnc/BasicSdncOperator.java
@@ -0,0 +1,148 @@
+/*-
+ * ============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 static org.junit.Assert.assertFalse;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.BiFunction;
+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.actor.test.BasicHttpOperation;
+import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
+import org.onap.policy.controlloop.actorserviceprovider.impl.HttpOperator;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
+import org.onap.policy.controlloop.policy.PolicyResult;
+import org.onap.policy.sdnc.SdncRequest;
+import org.onap.policy.sdnc.SdncResponse;
+import org.onap.policy.sdnc.SdncResponseOutput;
+import org.powermock.reflect.Whitebox;
+
+/**
+ * Superclass for various operator tests.
+ */
+public abstract class BasicSdncOperator extends BasicHttpOperation<SdncRequest> {
+
+ protected SdncResponse response;
+
+ /**
+ * Constructs the object using a default actor and operation name.
+ */
+ public BasicSdncOperator() {
+ super();
+ }
+
+ /**
+ * Constructs the object.
+ *
+ * @param actor actor name
+ * @param operation operation name
+ */
+ public BasicSdncOperator(String actor, String operation) {
+ super(actor, operation);
+ }
+
+ /**
+ * Initializes mocks and sets up.
+ */
+ public void setUp() throws Exception {
+ super.setUp();
+
+ response = new SdncResponse();
+
+ SdncResponseOutput output = new SdncResponseOutput();
+ response.setResponseOutput(output);
+ output.setResponseCode("200");
+
+ when(rawResponse.readEntity(String.class)).thenReturn(new StandardCoder().encode(response));
+ }
+
+ /**
+ * Runs the operation and verifies that the response is successful.
+ *
+ * @param operation operation to run
+ * @return the request that was posted
+ */
+ protected SdncRequest verifyOperation(SdncOperation operation)
+ throws InterruptedException, ExecutionException, TimeoutException {
+
+ CompletableFuture<OperationOutcome> future2 = operation.startOperationAsync(1, outcome);
+ assertFalse(future2.isDone());
+
+ verify(client).post(callbackCaptor.capture(), any(), requestCaptor.capture(), any());
+ callbackCaptor.getValue().completed(rawResponse);
+
+ assertEquals(PolicyResult.SUCCESS, future2.get(5, TimeUnit.SECONDS).getResult());
+
+ return requestCaptor.getValue().getEntity();
+ }
+
+ /**
+ * 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 fieldName name of the field to be removed from the enrichment data
+ * @param expectedText text expected in the exception message
+ */
+ protected void verifyMissing(String fieldName, String expectedText,
+ BiFunction<ControlLoopOperationParams,HttpOperator,SdncOperation> maker) {
+
+ makeContext();
+ enrichment.remove(fieldName);
+
+ SdncOperation oper = maker.apply(params, operator);
+
+ assertThatIllegalArgumentException().isThrownBy(() -> Whitebox.invokeMethod(oper, "makeRequest", 1))
+ .withMessageContaining("missing").withMessageContaining(expectedText);
+ }
+
+ 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
index 0a7bcad..7fc5ee7 100644
--- 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
@@ -26,45 +26,51 @@
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 {
+public class RerouteOperatorTest extends BasicSdncOperator {
- private RerouteOperator oper;
+ private RerouteOperation oper;
+ public RerouteOperatorTest() {
+ super(DEFAULT_ACTOR, RerouteOperation.NAME);
+ }
/**
* Set up.
*/
@Before
- public void setUp() {
- makeContext();
- oper = new RerouteOperator(ACTOR);
+ public void setUp() throws Exception {
+ super.setUp();
+ oper = new RerouteOperation(params, operator);
}
@Test
public void testRerouteOperator() {
- assertEquals(ACTOR, oper.getActorName());
- assertEquals(RerouteOperator.NAME, oper.getName());
+ assertEquals(DEFAULT_ACTOR, oper.getActorName());
+ assertEquals(RerouteOperation.NAME, oper.getName());
}
@Test
- public void testConstructRequest() throws CoderException {
- SdncRequest request = oper.constructRequest(context);
+ public void testMakeRequest() throws Exception {
+ SdncRequest request = oper.makeRequest(1);
assertEquals("my-service", request.getNsInstanceId());
assertEquals(REQ_ID, request.getRequestId());
- assertEquals(RerouteOperator.URI, request.getUrl());
+ assertEquals(RerouteOperation.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");
+ verifyMissing(RerouteOperation.SERVICE_ID_KEY, "service", RerouteOperation::new);
+ verifyMissing(RerouteOperation.NETWORK_ID_KEY, "network", RerouteOperation::new);
+
+ // perform the operation
+ makeContext();
+ verifyRequest("reroute.json", verifyOperation(oper));
}
@Override
protected Map<String, String> makeEnrichment() {
- return Map.of(RerouteOperator.SERVICE_ID_KEY, "my-service", RerouteOperator.NETWORK_ID_KEY, "my-network");
+ return Map.of(RerouteOperation.SERVICE_ID_KEY, "my-service", RerouteOperation.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 08655c3..ac81d49 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
@@ -41,7 +41,7 @@
public class SdncActorServiceProviderTest {
- private static final String REROUTE = RerouteOperator.NAME;
+ private static final String REROUTE = RerouteOperation.NAME;
/**
* Set up before test class.
@@ -63,7 +63,7 @@
final SdncActorServiceProvider prov = new SdncActorServiceProvider();
// verify that it has the operators we expect
- var expected = Arrays.asList(BandwidthOnDemandOperator.NAME, RerouteOperator.NAME).stream().sorted()
+ var expected = Arrays.asList(BandwidthOnDemandOperation.NAME, RerouteOperation.NAME).stream().sorted()
.collect(Collectors.toList());
var actual = prov.getOperationNames().stream().sorted().collect(Collectors.toList());
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
index 25d383e..4bc514c 100644
--- 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
@@ -21,306 +21,67 @@
package org.onap.policy.controlloop.actor.sdnc;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
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 java.util.TreeMap;
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";
+public class SdncOperatorTest extends BasicSdncOperator {
+
+ private SdncRequest request;
+ private SdncOperation oper;
/**
- * 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.
+ * Sets up.
*/
@Before
- public void setUp() {
- event = new VirtualControlLoopEvent();
- context = new ControlLoopEventContext(event);
+ public void setUp() throws Exception {
+ super.setUp();
- initOper(HTTP_CLIENT);
-
- output = new SdncResponseOutput();
- output.setResponseCode("200");
- }
-
- @After
- public void tearDown() {
- oper.shutdown();
+ oper = new SdncOperation(params, operator) {
+ @Override
+ protected SdncRequest makeRequest(int attempt) {
+ return request;
+ }
+ };
}
@Test
public void testSdncOperator() {
- assertEquals(ACTOR, oper.getActorName());
- assertEquals(OPERATION, oper.getName());
- assertEquals(ACTOR + "." + OPERATION, oper.getFullName());
+ assertEquals(DEFAULT_ACTOR, oper.getActorName());
+ assertEquals(DEFAULT_OPERATION, oper.getName());
}
@Test
- public void testGetClient() {
- assertNotNull(oper.getTheClient());
+ public void testStartOperationAsync_testStartRequestAsync() throws Exception {
+ verifyOperation(oper);
}
@Test
- public void testStartOperationAsync_testPostRequest() throws Exception {
- OperationOutcome outcome = runOperation();
- assertNotNull(outcome);
- assertEquals(PolicyResult.SUCCESS, outcome.getResult());
+ public void testIsSuccess() {
+ // success case
+ response.getResponseOutput().setResponseCode("200");
+ assertTrue(oper.isSuccess(null, response));
+
+ // failure code
+ response.getResponseOutput().setResponseCode("555");
+ assertFalse(oper.isSuccess(null, response));
+
+ // null code
+ response.getResponseOutput().setResponseCode(null);
+ assertFalse(oper.isSuccess(null, response));
+
+ // null output
+ response.setResponseOutput(null);
+ assertFalse(oper.isSuccess(null, response));
}
- /**
- * 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();
- }
+ @Override
+ protected Map<String, String> makeEnrichment() {
+ return new TreeMap<>();
}
}
diff --git a/models-interactions/model-actors/actor.test/pom.xml b/models-interactions/model-actors/actor.test/pom.xml
new file mode 100644
index 0000000..6b05807
--- /dev/null
+++ b/models-interactions/model-actors/actor.test/pom.xml
@@ -0,0 +1,69 @@
+<?xml version="1.0"?>
+<!--
+ ============LICENSE_START=======================================================
+ 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=========================================================
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.onap.policy.models.policy-models-interactions.model-actors</groupId>
+ <artifactId>model-actors</artifactId>
+ <version>2.2.1-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>actor.test</artifactId>
+ <description>Utilities for testing actors</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.onap.policy.models.policy-models-interactions.model-impl</groupId>
+ <artifactId>events</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.onap.policy.models.policy-models-interactions.model-actors</groupId>
+ <artifactId>actorServiceProvider</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.onap.policy.common</groupId>
+ <artifactId>policy-endpoints</artifactId>
+ <version>${policy.common.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.powermock</groupId>
+ <artifactId>powermock-api-mockito2</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/models-interactions/model-actors/actor.test/src/main/java/org/onap/policy/controlloop/actor/test/BasicHttpOperation.java b/models-interactions/model-actors/actor.test/src/main/java/org/onap/policy/controlloop/actor/test/BasicHttpOperation.java
new file mode 100644
index 0000000..15e4848
--- /dev/null
+++ b/models-interactions/model-actors/actor.test/src/main/java/org/onap/policy/controlloop/actor/test/BasicHttpOperation.java
@@ -0,0 +1,169 @@
+/*-
+ * ============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.test;
+
+import static org.mockito.Mockito.when;
+
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import javax.ws.rs.client.Entity;
+import javax.ws.rs.client.InvocationCallback;
+import javax.ws.rs.core.Response;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+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.VirtualControlLoopEvent;
+import org.onap.policy.controlloop.actorserviceprovider.ActorService;
+import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
+import org.onap.policy.controlloop.actorserviceprovider.controlloop.ControlLoopEventContext;
+import org.onap.policy.controlloop.actorserviceprovider.impl.HttpOperator;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
+
+/**
+ * Superclass for various operator tests.
+ *
+ * @param <Q> request type
+ */
+public class BasicHttpOperation<Q> {
+ protected static final UUID REQ_ID = UUID.randomUUID();
+ protected static final String DEFAULT_ACTOR = "default-actor";
+ protected static final String DEFAULT_OPERATION = "default-operation";
+ protected static final String MY_CLIENT = "my-client";
+ protected static final String BASE_URI = "/base-uri";
+ protected static final String PATH = "/my-path";
+ protected static final String TARGET_ENTITY = "my-target";
+
+ protected final String actorName;
+ protected final String operationName;
+
+ @Captor
+ protected ArgumentCaptor<InvocationCallback<Response>> callbackCaptor;
+
+ @Captor
+ protected ArgumentCaptor<Entity<Q>> requestCaptor;
+
+ @Captor
+ protected ArgumentCaptor<Map<String, Object>> headerCaptor;
+
+ @Mock
+ protected ActorService service;
+
+ @Mock
+ protected HttpClient client;
+
+ @Mock
+ protected HttpClientFactory factory;
+
+ @Mock
+ protected Response rawResponse;
+
+ @Mock
+ protected HttpOperator operator;
+
+ protected CompletableFuture<Response> future;
+ protected ControlLoopOperationParams params;
+ protected Map<String, String> enrichment;
+ protected VirtualControlLoopEvent event;
+ protected ControlLoopEventContext context;
+ protected OperationOutcome outcome;
+
+ /**
+ * Constructs the object using a default actor and operation name.
+ */
+ public BasicHttpOperation() {
+ this.actorName = DEFAULT_ACTOR;
+ this.operationName = DEFAULT_OPERATION;
+ }
+
+ /**
+ * Constructs the object.
+ *
+ * @param actor actor name
+ * @param operation operation name
+ */
+ public BasicHttpOperation(String actor, String operation) {
+ this.actorName = actor;
+ this.operationName = operation;
+ }
+
+ /**
+ * Initializes mocks and sets up.
+ */
+ public void setUp() throws Exception {
+ MockitoAnnotations.initMocks(this);
+
+ when(factory.get(MY_CLIENT)).thenReturn(client);
+
+ when(rawResponse.getStatus()).thenReturn(200);
+
+ future = new CompletableFuture<>();
+ when(client.getBaseUrl()).thenReturn(BASE_URI);
+
+ makeContext();
+
+ outcome = params.makeOutcome();
+
+ initOperator();
+ }
+
+ /**
+ * Reinitializes {@link #enrichment}, {@link #event}, {@link #context}, and
+ * {@link #params}.
+ */
+ protected void makeContext() {
+ enrichment = new TreeMap<>(makeEnrichment());
+
+ event = new VirtualControlLoopEvent();
+ event.setRequestId(REQ_ID);
+ event.setAai(enrichment);
+
+ context = new ControlLoopEventContext(event);
+
+ params = ControlLoopOperationParams.builder().context(context).actorService(service).actor(actorName)
+ .operation(operationName).targetEntity(TARGET_ENTITY).build();
+ }
+
+ /**
+ * Initializes an operator so that it is "alive" and has the given names.
+ */
+ protected void initOperator() {
+ when(operator.isAlive()).thenReturn(true);
+ when(operator.getFullName()).thenReturn(actorName + "." + operationName);
+ when(operator.getActorName()).thenReturn(actorName);
+ when(operator.getName()).thenReturn(operationName);
+ when(operator.getClient()).thenReturn(client);
+ when(operator.getPath()).thenReturn(PATH);
+ }
+
+ /**
+ * Makes enrichment data.
+ *
+ * @return enrichment data
+ */
+ protected Map<String, String> makeEnrichment() {
+ return new TreeMap<>();
+ }
+}
diff --git a/models-interactions/model-actors/actor.test/src/test/java/org/onap/policy/controlloop/actor/test/BasicHttpOperationTest.java b/models-interactions/model-actors/actor.test/src/test/java/org/onap/policy/controlloop/actor/test/BasicHttpOperationTest.java
new file mode 100644
index 0000000..c33483d
--- /dev/null
+++ b/models-interactions/model-actors/actor.test/src/test/java/org/onap/policy/controlloop/actor/test/BasicHttpOperationTest.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.actor.test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Before;
+import org.junit.Test;
+
+public class BasicHttpOperationTest {
+ private static final String ACTOR = "my-actor";
+ private static final String OPERATION = "my-operation";
+
+ private BasicHttpOperation<String> oper;
+
+
+ @Before
+ public void setUp() throws Exception {
+ oper = new BasicHttpOperation<>(ACTOR, OPERATION);
+ oper.setUp();
+ }
+
+ @Test
+ public void testBasicHttpOperation() {
+ oper = new BasicHttpOperation<>();
+ assertEquals(BasicHttpOperation.DEFAULT_ACTOR, oper.actorName);
+ assertEquals(BasicHttpOperation.DEFAULT_OPERATION, oper.operationName);
+ }
+
+ @Test
+ public void testBasicHttpOperationStringString() {
+ assertEquals(ACTOR, oper.actorName);
+ assertEquals(OPERATION, oper.operationName);
+ }
+
+ @Test
+ public void testSetUp() {
+ assertNotNull(oper.client);
+ assertSame(oper.client, oper.factory.get(BasicHttpOperation.MY_CLIENT));
+ assertEquals(200, oper.rawResponse.getStatus());
+ assertNotNull(oper.future);
+ assertEquals(BasicHttpOperation.BASE_URI, oper.client.getBaseUrl());
+ assertNotNull(oper.context);
+ assertNotNull(oper.outcome);
+ assertTrue(oper.operator.isAlive());
+ }
+
+ @Test
+ public void testMakeContext() {
+ oper.makeContext();
+
+ assertTrue(oper.enrichment.isEmpty());
+
+ assertSame(BasicHttpOperation.REQ_ID, oper.event.getRequestId());
+ assertSame(oper.enrichment, oper.event.getAai());
+
+ assertSame(oper.event, oper.context.getEvent());
+
+ assertSame(oper.context, oper.params.getContext());
+ assertSame(oper.service, oper.params.getActorService());
+ assertEquals(ACTOR, oper.params.getActor());
+ assertEquals(OPERATION, oper.params.getOperation());
+ assertEquals(BasicHttpOperation.TARGET_ENTITY, oper.params.getTargetEntity());
+ }
+
+ @Test
+ public void testInitOperator() throws Exception {
+ oper.initOperator();
+
+ assertTrue(oper.operator.isAlive());
+ assertEquals(ACTOR + "." + OPERATION, oper.operator.getFullName());
+ assertEquals(ACTOR, oper.operator.getActorName());
+ assertEquals(OPERATION, oper.operator.getName());
+ assertSame(oper.client, oper.operator.getClient());
+ assertEquals(BasicHttpOperation.PATH, oper.operator.getPath());
+ }
+
+ @Test
+ public void testMakeEnrichment() {
+ assertTrue(oper.makeEnrichment().isEmpty());
+ }
+
+}
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
deleted file mode 100644
index d784038..0000000
--- a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/AsyncResponseHandler.java
+++ /dev/null
@@ -1,119 +0,0 @@
-/*-
- * ============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/Operation.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/Operation.java
new file mode 100644
index 0000000..39977fd
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/Operation.java
@@ -0,0 +1,52 @@
+/*-
+ * ============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;
+
+/**
+ * This is the service interface for defining an Actor operation used in Control Loop
+ * Operational Policies for performing actions on runtime entities.
+ */
+public interface Operation {
+
+ /**
+ * Gets the name of the associated actor.
+ *
+ * @return the name of the associated actor
+ */
+ String getActorName();
+
+ /**
+ * Gets the name of the operation.
+ *
+ * @return the operation name
+ */
+ String getName();
+
+ /**
+ * Called by enforcement PDP engine to start the operation. As part of the operation,
+ * it invokes the "start" and "complete" call-backs found within the parameters.
+ *
+ * @return a future that can be used to cancel or await the result of the operation
+ */
+ CompletableFuture<OperationOutcome> start();
+}
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 c09460e..24faafd 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
@@ -21,7 +21,6 @@
package org.onap.policy.controlloop.actorserviceprovider;
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.actorserviceprovider.parameters.ControlLoopOperationParams;
@@ -47,11 +46,10 @@
String getName();
/**
- * Called by enforcement PDP engine to start the operation. As part of the operation,
- * it invokes the "start" and "complete" call-backs found within the parameters.
+ * Called by enforcement PDP engine to build the operation.
*
- * @param params parameters needed to start the operation
- * @return a future that can be used to cancel or await the result of the operation
+ * @param params parameters needed by the operation
+ * @return a new operation
*/
- CompletableFuture<OperationOutcome> startOperation(ControlLoopOperationParams params);
+ Operation buildOperation(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 c3ddd17..b885b5c 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,9 +23,6 @@
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;
@@ -56,82 +53,6 @@
}
/**
- * 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.
*
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 cd4d257..1c37a8e 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
@@ -23,12 +23,15 @@
import java.io.Serializable;
import java.util.Map;
import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import lombok.AccessLevel;
import lombok.Getter;
import lombok.NonNull;
import lombok.Setter;
import org.onap.policy.controlloop.VirtualControlLoopEvent;
+import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
/**
* Context associated with a control loop event.
@@ -47,11 +50,23 @@
*/
private final Map<String, String> enrichment;
+ /**
+ * Set of properties that have been stored in the context.
+ */
@Getter(AccessLevel.NONE)
@Setter(AccessLevel.NONE)
private Map<String, Serializable> properties = new ConcurrentHashMap<>();
/**
+ * When {@link #obtain(String, ControlLoopOperationParams)} is invoked and the
+ * specified property is not found in {@link #properties}, it is retrieved. This holds
+ * the futures for the operations retrieving the properties.
+ */
+ @Getter(AccessLevel.NONE)
+ @Setter(AccessLevel.NONE)
+ private transient Map<String, CompletableFuture<OperationOutcome>> retrievers = new ConcurrentHashMap<>();
+
+ /**
* Request ID extracted from the event, or a generated value if the event has no
* request id; never {@code null}.
*/
@@ -100,4 +115,34 @@
public void setProperty(String name, Serializable value) {
properties.put(name, value);
}
+
+ /**
+ * Obtains the given property.
+ *
+ * @param name name of the desired property
+ * @param params parameters needed to perform the operation to retrieve the desired
+ * property
+ * @return a future for retrieving the property, {@code null} if the property has
+ * already been retrieved
+ */
+ public CompletableFuture<OperationOutcome> obtain(String name, ControlLoopOperationParams params) {
+ if (properties.containsKey(name)) {
+ return null;
+ }
+
+ CompletableFuture<OperationOutcome> future = retrievers.get(name);
+ if (future != null) {
+ return future;
+ }
+
+ future = params.start();
+
+ CompletableFuture<OperationOutcome> oldFuture = retrievers.putIfAbsent(name, future);
+ if (oldFuture != null) {
+ future.cancel(false);
+ return oldFuture;
+ }
+
+ return future;
+ }
}
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 d7f322e..0c88ebe 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
@@ -91,7 +91,7 @@
public Operator getOperator(String name) {
Operator operator = name2operator.get(name);
if (operator == null) {
- throw new IllegalArgumentException("unknown operation " + getName() + "." + name);
+ throw new IllegalArgumentException("unknown operator " + getName() + "." + name);
}
return operator;
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperation.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperation.java
new file mode 100644
index 0000000..c4bf5f4
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperation.java
@@ -0,0 +1,286 @@
+/*-
+ * ============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.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Future;
+import java.util.function.Function;
+import javax.ws.rs.client.InvocationCallback;
+import javax.ws.rs.core.Response;
+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.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;
+import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.HttpParams;
+import org.onap.policy.controlloop.actorserviceprovider.pipeline.PipelineControllerFuture;
+import org.onap.policy.controlloop.policy.PolicyResult;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Operator that uses HTTP. The operator's parameters must be an {@link HttpParams}.
+ *
+ * @param <T> response type
+ */
+@Getter
+public abstract class HttpOperation<T> extends OperationPartial {
+ private static final Logger logger = LoggerFactory.getLogger(HttpOperation.class);
+ private static final Coder coder = new StandardCoder();
+
+ /**
+ * Operator that created this operation.
+ */
+ protected final HttpOperator operator;
+
+ /**
+ * Response class.
+ */
+ private final Class<T> responseClass;
+
+
+ /**
+ * Constructs the object.
+ *
+ * @param params operation parameters
+ * @param operator operator that created this operation
+ * @param clazz response class
+ */
+ public HttpOperation(ControlLoopOperationParams params, HttpOperator operator, Class<T> clazz) {
+ super(params, operator);
+ this.operator = operator;
+ this.responseClass = clazz;
+ }
+
+ /**
+ * If no timeout is specified, then it returns the operator's configured timeout.
+ */
+ @Override
+ protected long getTimeoutMs(Integer timeoutSec) {
+ return (timeoutSec == null || timeoutSec == 0 ? operator.getTimeoutMs() : super.getTimeoutMs(timeoutSec));
+ }
+
+ /**
+ * Makes the request headers. This simply returns an empty map.
+ *
+ * @return request headers, a non-null, modifiable map
+ */
+ protected Map<String, Object> makeHeaders() {
+ return new HashMap<>();
+ }
+
+ /**
+ * Gets the path to be used when performing the request; this is typically appended to
+ * the base URL. This method simply invokes {@link #getPath()}.
+ *
+ * @return the path URI suffix
+ */
+ public String makePath() {
+ return operator.getPath();
+ }
+
+ /**
+ * Makes the URL to which the "get" request should be posted. This ir primarily used
+ * for logging purposes. This particular method returns the base URL appended with the
+ * return value from {@link #makePath()}.
+ *
+ * @return the URL to which from which to get
+ */
+ public String makeUrl() {
+ return (operator.getClient().getBaseUrl() + makePath());
+ }
+
+ /**
+ * Arranges to handle a response.
+ *
+ * @param outcome outcome to be populate
+ * @param url URL to which to request was sent
+ * @param requester function to initiate the request and invoke the given callback
+ * when it completes
+ * @return a future for the response
+ */
+ protected CompletableFuture<OperationOutcome> handleResponse(OperationOutcome outcome, String url,
+ Function<InvocationCallback<Response>, Future<Response>> requester) {
+
+ final PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+ final CompletableFuture<Response> future = new CompletableFuture<>();
+ final Executor executor = params.getExecutor();
+
+ // arrange for the callback to complete "future"
+ InvocationCallback<Response> callback = new InvocationCallback<>() {
+ @Override
+ public void completed(Response response) {
+ future.complete(response);
+ }
+
+ @Override
+ public void failed(Throwable throwable) {
+ logger.warn("{}.{}: response failure for {}", params.getActor(), params.getOperation(),
+ params.getRequestId());
+ future.completeExceptionally(throwable);
+ }
+ };
+
+ // start the request and arrange to cancel it if the controller is canceled
+ controller.add(requester.apply(callback));
+
+ // once "future" completes, process the response, and then complete the controller
+ future.thenApplyAsync(response -> processResponse(outcome, url, response), executor)
+ .whenCompleteAsync(controller.delayedComplete(), executor);
+
+ return controller;
+ }
+
+ /**
+ * Processes a response. This method simply sets the outcome to SUCCESS.
+ *
+ * @param outcome outcome to be populate
+ * @param url URL to which to request was sent
+ * @param response raw response to process
+ * @return the outcome
+ */
+ protected OperationOutcome processResponse(OperationOutcome outcome, String url, Response rawResponse) {
+
+ logger.info("{}.{}: response received for {}", params.getActor(), params.getOperation(), params.getRequestId());
+
+ String strResponse = HttpClient.getBody(rawResponse, String.class);
+
+ logRestResponse(url, strResponse);
+
+ T response;
+ if (responseClass == String.class) {
+ response = responseClass.cast(strResponse);
+
+ } else {
+ try {
+ response = makeCoder().decode(strResponse, responseClass);
+ } catch (CoderException e) {
+ logger.warn("{}.{} cannot decode response with http error code {} for {}", params.getActor(),
+ params.getOperation(), rawResponse.getStatus(), params.getRequestId(), e);
+ return setOutcome(outcome, PolicyResult.FAILURE_EXCEPTION);
+ }
+ }
+
+ if (!isSuccess(rawResponse, response)) {
+ logger.info("{}.{} request failed with http error code {} for {}", params.getActor(), params.getOperation(),
+ rawResponse.getStatus(), params.getRequestId());
+ return setOutcome(outcome, PolicyResult.FAILURE);
+ }
+
+ logger.info("{}.{} request succeeded for {}", params.getActor(), params.getOperation(), params.getRequestId());
+ setOutcome(outcome, PolicyResult.SUCCESS);
+ postProcessResponse(outcome, url, rawResponse, response);
+
+ return outcome;
+ }
+
+ /**
+ * Processes a successful response.
+ *
+ * @param outcome outcome to be populate
+ * @param url URL to which to request was sent
+ * @param rawResponse raw response
+ * @param response decoded response
+ */
+ protected void postProcessResponse(OperationOutcome outcome, String url, Response rawResponse, T response) {
+ // do nothing
+ }
+
+ /**
+ * Determines if the response indicates success. This method simply checks the HTTP
+ * status code.
+ *
+ * @param rawResponse raw response
+ * @param response decoded response
+ * @return {@code true} if the response indicates success, {@code false} otherwise
+ */
+ protected boolean isSuccess(Response rawResponse, T response) {
+ return (rawResponse.getStatus() == 200);
+ }
+
+ /**
+ * 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 <Q> void logRestRequest(String url, Q request) {
+ String json;
+ try {
+ if (request == null) {
+ json = null;
+ } else if (request instanceof String) {
+ json = request.toString();
+ } else {
+ json = makeCoder().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 <S> void logRestResponse(String url, S response) {
+ String json;
+ try {
+ if (response == null) {
+ json = null;
+ } else if (response instanceof String) {
+ json = response.toString();
+ } else {
+ json = makeCoder().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);
+ }
+
+ // these may be overridden by junit tests
+
+ protected Coder makeCoder() {
+ return coder;
+ }
+}
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
index 5664929..add74aa 100644
--- 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
@@ -21,31 +21,35 @@
package org.onap.policy.controlloop.actorserviceprovider.impl;
import java.util.Map;
-import lombok.AccessLevel;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiFunction;
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.Operation;
import org.onap.policy.controlloop.actorserviceprovider.Util;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
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}.
+ * Operator that uses HTTP. The operator's parameters must be an {@link HttpParams}.
*/
-public class HttpOperator extends OperatorPartial {
+@Getter
+public abstract class HttpOperator extends OperatorPartial {
- @Getter(AccessLevel.PROTECTED)
private HttpClient client;
- @Getter
- private long timeoutSec;
+ /**
+ * Default timeout, in milliseconds, if none specified in the request.
+ */
+ private long timeoutMs;
/**
- * URI path for this particular operation.
+ * URI path for this particular operation. Includes a leading "/".
*/
- @Getter
private String path;
@@ -60,6 +64,26 @@
}
/**
+ * Makes an operator that will construct operations.
+ *
+ * @param <T> response type
+ * @param actorName actor name
+ * @param operation operation name
+ * @param operationMaker function to make an operation
+ * @return a new operator
+ */
+ public static <T> HttpOperator makeOperator(String actorName, String operation,
+ BiFunction<ControlLoopOperationParams, HttpOperator, HttpOperation<T>> operationMaker) {
+
+ return new HttpOperator(actorName, operation) {
+ @Override
+ public Operation buildOperation(ControlLoopOperationParams params) {
+ return operationMaker.apply(params, this);
+ }
+ };
+ }
+
+ /**
* Translates the parameters to an {@link HttpParams} and then extracts the relevant
* values.
*/
@@ -73,10 +97,10 @@
client = getClientFactory().get(params.getClientName());
path = params.getPath();
- timeoutSec = params.getTimeoutSec();
+ timeoutMs = TimeUnit.MILLISECONDS.convert(params.getTimeoutSec(), TimeUnit.SECONDS);
}
- // these may be overridden by junits
+ // these may be overridden by junit tests
protected HttpClientFactory getClientFactory() {
return HttpClientFactoryInstance.getClientFactory();
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperationPartial.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperationPartial.java
new file mode 100644
index 0000000..d00b88b
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperationPartial.java
@@ -0,0 +1,844 @@
+/*-
+ * ============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.List;
+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.function.BiConsumer;
+import java.util.function.Function;
+import org.onap.policy.controlloop.ControlLoopOperation;
+import org.onap.policy.controlloop.actorserviceprovider.CallbackManager;
+import org.onap.policy.controlloop.actorserviceprovider.Operation;
+import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
+import org.onap.policy.controlloop.actorserviceprovider.pipeline.PipelineControllerFuture;
+import org.onap.policy.controlloop.policy.PolicyResult;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Partial implementation of an operator. In general, it's preferable that subclasses
+ * would override {@link #startOperationAsync(int, OperationOutcome)
+ * startOperationAsync()}. However, if that proves to be too difficult, then they can
+ * simply override {@link #doOperation(int, OperationOutcome) doOperation()}. In addition,
+ * if the operation requires any preprocessor steps, the subclass may choose to override
+ * {@link #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(int, OperationOutcome) doOperation()}, then there's little that can
+ * be done to cancel that particular operation.
+ */
+public abstract class OperationPartial implements Operation {
+
+ private static final Logger logger = LoggerFactory.getLogger(OperationPartial.class);
+ public static final long DEFAULT_RETRY_WAIT_MS = 1000L;
+
+ // values extracted from the operator
+
+ private final OperatorPartial operator;
+
+ /**
+ * Operation parameters.
+ */
+ protected final ControlLoopOperationParams params;
+
+
+ /**
+ * Constructs the object.
+ *
+ * @param params operation parameters
+ * @param operator operator that created this operation
+ */
+ public OperationPartial(ControlLoopOperationParams params, OperatorPartial operator) {
+ this.params = params;
+ this.operator = operator;
+ }
+
+ public Executor getBlockingExecutor() {
+ return operator.getBlockingExecutor();
+ }
+
+ public String getFullName() {
+ return operator.getFullName();
+ }
+
+ public String getActorName() {
+ return operator.getActorName();
+ }
+
+ public String getName() {
+ return operator.getName();
+ }
+
+ @Override
+ public final CompletableFuture<OperationOutcome> start() {
+ if (!operator.isAlive()) {
+ throw new IllegalStateException("operation is not running: " + getFullName());
+ }
+
+ // allocate a controller for the entire operation
+ final PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+
+ CompletableFuture<OperationOutcome> preproc = startPreprocessorAsync();
+ if (preproc == null) {
+ // no preprocessor required - just start the operation
+ return startOperationAttempt(controller, 1);
+ }
+
+ /*
+ * 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.
+ */
+ // @formatter:off
+ controller.wrap(preproc)
+ .exceptionally(fromException("preprocessor of operation"))
+ .thenCompose(handlePreprocessorFailure(controller))
+ .thenCompose(unusedOutcome -> startOperationAttempt(controller, 1))
+ .whenCompleteAsync(controller.delayedComplete(), params.getExecutor());
+ // @formatter:on
+
+ return controller;
+ }
+
+ /**
+ * Handles a failure in the preprocessor pipeline. If a failure occurred, then it
+ * 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 controller pipeline controller
+ * @return a function that checks the outcome status and continues, if successful, or
+ * indicates a failure otherwise
+ */
+ private Function<OperationOutcome, CompletableFuture<OperationOutcome>> handlePreprocessorFailure(
+ PipelineControllerFuture<OperationOutcome> controller) {
+
+ return outcome -> {
+
+ if (outcome != null && isSuccess(outcome)) {
+ logger.info("{}: preprocessor succeeded for {}", getFullName(), params.getRequestId());
+ return CompletableFuture.completedFuture(outcome);
+ }
+
+ logger.warn("preprocessor failed, discontinuing operation {} for {}", getFullName(), params.getRequestId());
+
+ final Executor executor = params.getExecutor();
+ final CallbackManager callbacks = new CallbackManager();
+
+ // propagate "stop" to the callbacks
+ controller.add(callbacks);
+
+ final OperationOutcome outcome2 = params.makeOutcome();
+
+ // TODO need a FAILURE_MISSING_DATA (e.g., A&AI)
+
+ outcome2.setResult(PolicyResult.FAILURE_GUARD);
+ outcome2.setMessage(outcome != null ? outcome.getMessage() : null);
+
+ // @formatter:off
+ CompletableFuture.completedFuture(outcome2)
+ .whenCompleteAsync(callbackStarted(callbacks), executor)
+ .whenCompleteAsync(callbackCompleted(callbacks), executor)
+ .whenCompleteAsync(controller.delayedComplete(), executor);
+ // @formatter:on
+
+ return new CompletableFuture<>();
+ };
+ }
+
+ /**
+ * Invokes the operation's preprocessor step(s) as a "future". This method simply
+ * invokes {@link #startGuardAsync()}.
+ * <p/>
+ * This method assumes the following:
+ * <ul>
+ * <li>the operator is alive</li>
+ * <li>exceptions generated within the pipeline will be handled by the invoker</li>
+ * </ul>
+ *
+ * @return a function that will start the preprocessor and returns its outcome, or
+ * {@code null} if this operation needs no preprocessor
+ */
+ protected CompletableFuture<OperationOutcome> startPreprocessorAsync() {
+ return startGuardAsync();
+ }
+
+ /**
+ * Invokes the operation's guard step(s) as a "future". This method simply returns
+ * {@code null}.
+ * <p/>
+ * This method assumes the following:
+ * <ul>
+ * <li>the operator is alive</li>
+ * <li>exceptions generated within the pipeline will be handled by the invoker</li>
+ * </ul>
+ *
+ * @return a function that will start the guard checks and returns its outcome, or
+ * {@code null} if this operation has no guard
+ */
+ protected CompletableFuture<OperationOutcome> startGuardAsync() {
+ return null;
+ }
+
+ /**
+ * Starts the operation attempt, with no preprocessor. When all retries complete, it
+ * will complete the controller.
+ *
+ * @param controller controller for all operation attempts
+ * @param attempt attempt number, typically starting with 1
+ * @return a future that will return the final result of all attempts
+ */
+ private CompletableFuture<OperationOutcome> startOperationAttempt(
+ PipelineControllerFuture<OperationOutcome> controller, int attempt) {
+
+ // propagate "stop" to the operation attempt
+ controller.wrap(startAttemptWithoutRetries(attempt)).thenCompose(retryOnFailure(controller, attempt))
+ .whenCompleteAsync(controller.delayedComplete(), params.getExecutor());
+
+ return controller;
+ }
+
+ /**
+ * Starts the operation attempt, without doing any retries.
+ *
+ * @param params operation parameters
+ * @param attempt attempt number, typically starting with 1
+ * @return a future that will return the result of a single operation attempt
+ */
+ private CompletableFuture<OperationOutcome> startAttemptWithoutRetries(int attempt) {
+
+ logger.info("{}: start operation attempt {} for {}", getFullName(), attempt, params.getRequestId());
+
+ final Executor executor = params.getExecutor();
+ final OperationOutcome outcome = params.makeOutcome();
+ final CallbackManager callbacks = new CallbackManager();
+
+ // this operation attempt gets its own controller
+ final PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+
+ // propagate "stop" to the callbacks
+ controller.add(callbacks);
+
+ // @formatter:off
+ CompletableFuture<OperationOutcome> future = CompletableFuture.completedFuture(outcome)
+ .whenCompleteAsync(callbackStarted(callbacks), executor)
+ .thenCompose(controller.wrap(outcome2 -> startOperationAsync(attempt, outcome2)));
+ // @formatter:on
+
+ // handle timeouts, if specified
+ long timeoutMillis = getTimeoutMs(params.getTimeoutSec());
+ if (timeoutMillis > 0) {
+ logger.info("{}: set timeout to {}ms for {}", getFullName(), timeoutMillis, params.getRequestId());
+ future = future.orTimeout(timeoutMillis, TimeUnit.MILLISECONDS);
+ }
+
+ /*
+ * Note: we re-invoke callbackStarted() just to be sure the callback is invoked
+ * before callbackCompleted() is invoked.
+ *
+ * Note: no need to remove "callbacks" from the pipeline, as we're going to stop
+ * the pipeline as the last step anyway.
+ */
+
+ // @formatter:off
+ future.exceptionally(fromException("operation"))
+ .thenApply(setRetryFlag(attempt))
+ .whenCompleteAsync(callbackStarted(callbacks), executor)
+ .whenCompleteAsync(callbackCompleted(callbacks), executor)
+ .whenCompleteAsync(controller.delayedComplete(), executor);
+ // @formatter:on
+
+ return controller;
+ }
+
+ /**
+ * Determines if the outcome was successful.
+ *
+ * @param outcome outcome to examine
+ * @return {@code true} if the outcome was successful
+ */
+ protected boolean isSuccess(OperationOutcome outcome) {
+ return (outcome.getResult() == PolicyResult.SUCCESS);
+ }
+
+ /**
+ * Determines if the outcome was a failure for this operator.
+ *
+ * @param outcome outcome to examine, or {@code null}
+ * @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(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()} 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>
+ * <li>the operator is alive</li>
+ * <li>verifyRunning() has been invoked</li>
+ * <li>callbackStarted() has been invoked</li>
+ * <li>the invoker will perform appropriate timeout checks</li>
+ * <li>exceptions generated within the pipeline will be handled by the invoker</li>
+ * </ul>
+ *
+ * @param attempt attempt number, typically starting with 1
+ * @return a function that will start the operation and return its result when
+ * complete
+ */
+ protected CompletableFuture<OperationOutcome> startOperationAsync(int attempt, OperationOutcome outcome) {
+
+ return CompletableFuture.supplyAsync(() -> doOperation(attempt, outcome), getBlockingExecutor());
+ }
+
+ /**
+ * Low-level method that performs the operation. This can make the same assumptions
+ * that are made by {@link #doOperationAsFuture()}. This particular method simply
+ * throws an {@link UnsupportedOperationException}.
+ *
+ * @param attempt attempt number, typically starting with 1
+ * @param operation the operation being performed
+ * @return the outcome of the operation
+ */
+ protected OperationOutcome doOperation(int attempt, OperationOutcome operation) {
+
+ throw new UnsupportedOperationException("start operation " + getFullName());
+ }
+
+ /**
+ * Sets the outcome status to FAILURE_RETRIES, if the current operation outcome is
+ * FAILURE, assuming the policy specifies retries and the retry count has been
+ * exhausted.
+ *
+ * @param attempt latest attempt number, starting with 1
+ * @return a function to get the next future to execute
+ */
+ private Function<OperationOutcome, OperationOutcome> setRetryFlag(int attempt) {
+
+ return operation -> {
+ if (operation != null && !isActorFailed(operation)) {
+ /*
+ * wrong type or wrong operation - just leave it as is. No need to log
+ * anything here, as retryOnFailure() will log a message
+ */
+ return operation;
+ }
+
+ // get a non-null operation
+ OperationOutcome oper2;
+ if (operation != null) {
+ oper2 = operation;
+ } else {
+ oper2 = params.makeOutcome();
+ oper2.setResult(PolicyResult.FAILURE);
+ }
+
+ int retry = getRetry(params.getRetry());
+ if (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.setResult(PolicyResult.FAILURE_RETRIES);
+ }
+
+ return oper2;
+ };
+ }
+
+ /**
+ * Restarts the operation if it was a FAILURE. Assumes that {@link #setRetryFlag(int)}
+ * was previously invoked, and thus that the "operation" is not {@code null}.
+ *
+ * @param controller controller for all of the retries
+ * @param attempt latest attempt number, starting with 1
+ * @return a function to get the next future to execute
+ */
+ private Function<OperationOutcome, CompletableFuture<OperationOutcome>> retryOnFailure(
+ PipelineControllerFuture<OperationOutcome> controller, int attempt) {
+
+ return operation -> {
+ if (!isActorFailed(operation)) {
+ // wrong type or wrong operation - just leave it as is
+ logger.info("not retrying operation {} for {}", getFullName(), params.getRequestId());
+ controller.complete(operation);
+ return new CompletableFuture<>();
+ }
+
+ if (getRetry(params.getRetry()) <= 0) {
+ // no retries - already marked as FAILURE, so just return it
+ logger.info("operation {} no retries for {}", getFullName(), params.getRequestId());
+ controller.complete(operation);
+ return new CompletableFuture<>();
+ }
+
+ /*
+ * Retry the operation.
+ */
+ long waitMs = getRetryWaitMs();
+ logger.info("retry operation {} in {}ms for {}", getFullName(), waitMs, params.getRequestId());
+
+ return sleep(waitMs, TimeUnit.MILLISECONDS)
+ .thenCompose(unused -> startOperationAttempt(controller, attempt + 1));
+ };
+ }
+
+ /**
+ * Convenience method that starts a sleep(), running via a future.
+ *
+ * @param sleepTime time to sleep
+ * @param unit time unit
+ * @return a future that will complete when the sleep completes
+ */
+ protected CompletableFuture<Void> sleep(long sleepTime, TimeUnit unit) {
+ if (sleepTime <= 0) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ return new CompletableFuture<Void>().completeOnTimeout(null, sleepTime, unit);
+ }
+
+ /**
+ * Converts an exception into an operation outcome, returning a copy of the outcome to
+ * prevent background jobs from changing it.
+ *
+ * @param type type of item throwing the exception
+ * @return a function that will convert an exception into an operation outcome
+ */
+ private Function<Throwable, OperationOutcome> fromException(String type) {
+
+ return thrown -> {
+ OperationOutcome outcome = params.makeOutcome();
+
+ logger.warn("exception throw by {} {}.{} for {}", type, outcome.getActor(), outcome.getOperation(),
+ params.getRequestId(), thrown);
+
+ return setOutcome(outcome, thrown);
+ };
+ }
+
+ /**
+ * Similar to {@link CompletableFuture#anyOf(CompletableFuture...)}, but it cancels
+ * any outstanding futures when one completes.
+ *
+ * @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(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(arrFutures);
+ return result;
+ }
+
+ /**
+ * Same as {@link CompletableFuture#anyOf(CompletableFuture...)}, but it cancels any
+ * outstanding futures when one completes.
+ *
+ * @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(
+ @SuppressWarnings("unchecked") CompletableFuture<OperationOutcome>... futures) {
+
+ if (futures.length == 1) {
+ return futures[0];
+ }
+
+ 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 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(List<CompletableFuture<OperationOutcome>> futures) {
+
+ // convert list to an array
+ @SuppressWarnings("rawtypes")
+ CompletableFuture[] arrFutures = futures.toArray(new CompletableFuture[futures.size()]);
+
+ @SuppressWarnings("unchecked")
+ CompletableFuture<OperationOutcome> result = allOf(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 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(
+ @SuppressWarnings("unchecked") CompletableFuture<OperationOutcome>... futures) {
+
+ if (futures.length == 1) {
+ return futures[0];
+ }
+
+ 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);
+ }
+
+ // @formatter:off
+ CompletableFuture.allOf(futures2)
+ .thenApply(unused -> combineOutcomes(outcomes))
+ .whenCompleteAsync(controller.delayedComplete(), params.getExecutor());
+ // @formatter:on
+
+ return controller;
+ }
+
+ /**
+ * 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) {
+
+ if (futures.length == 0) {
+ throw new IllegalArgumentException("empty list of futures");
+ }
+
+ // attach each task
+ for (CompletableFuture<OperationOutcome> future : futures) {
+ controller.add(future);
+ }
+ }
+
+ /**
+ * Combines the outcomes from a set of tasks.
+ *
+ * @param outcomes outcomes to be examined
+ * @return the combined outcome
+ */
+ private OperationOutcome combineOutcomes(OperationOutcome[] outcomes) {
+
+ // identify the outcome with the highest priority
+ OperationOutcome outcome = outcomes[0];
+ int priority = detmPriority(outcome);
+
+ // 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);
+
+ if (priority2 > priority) {
+ outcome = outcome2;
+ priority = priority2;
+ }
+ }
+
+ logger.info("{}: combined outcome of tasks is {} for {}", getFullName(),
+ (outcome == null ? null : outcome.getResult()), params.getRequestId());
+
+ return outcome;
+ }
+
+ /**
+ * Determines the priority of an outcome based on its result.
+ *
+ * @param outcome outcome to examine, or {@code null}
+ * @return the outcome's priority
+ */
+ protected int detmPriority(OperationOutcome outcome) {
+ if (outcome == null || outcome.getResult() == null) {
+ return 1;
+ }
+
+ 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 controller overall pipeline controller
+ * @param checkSuccess {@code true} to check the previous outcome, {@code false}
+ * otherwise
+ * @param outcome outcome of the previous task
+ * @param task task to be performed
+ * @return the task, if everything checks out. Otherwise, it returns an incomplete
+ * future and completes the controller instead
+ */
+ // @formatter:off
+ protected CompletableFuture<OperationOutcome> doTask(
+ PipelineControllerFuture<OperationOutcome> controller,
+ boolean checkSuccess, OperationOutcome outcome,
+ CompletableFuture<OperationOutcome> task) {
+ // @formatter:on
+
+ if (checkSuccess && !isSuccess(outcome)) {
+ /*
+ * must complete before canceling so that cancel() doesn't cause controller to
+ * complete
+ */
+ controller.complete(outcome);
+ task.cancel(false);
+ return new CompletableFuture<>();
+ }
+
+ return controller.wrap(task);
+ }
+
+ /**
+ * Performs a task, after verifying that the controller is still running. Also checks
+ * that the previous outcome was successful, if specified.
+ *
+ * @param controller overall pipeline controller
+ * @param checkSuccess {@code true} to check the previous outcome, {@code false}
+ * otherwise
+ * @param task function to start the task to be performed
+ * @return a function to perform the task. If everything checks out, then it returns
+ * the task. Otherwise, it returns an incomplete future and completes the
+ * controller instead
+ */
+ // @formatter:off
+ protected Function<OperationOutcome, CompletableFuture<OperationOutcome>> doTask(
+ PipelineControllerFuture<OperationOutcome> controller,
+ boolean checkSuccess,
+ Function<OperationOutcome, CompletableFuture<OperationOutcome>> task) {
+ // @formatter:on
+
+ return outcome -> {
+
+ if (!controller.isRunning()) {
+ return new CompletableFuture<>();
+ }
+
+ if (checkSuccess && !isSuccess(outcome)) {
+ controller.complete(outcome);
+ return new CompletableFuture<>();
+ }
+
+ return controller.wrap(task.apply(outcome));
+ };
+ }
+
+ /**
+ * Sets the start time of the operation and invokes the callback to indicate that the
+ * operation has started. Does nothing if the pipeline has been stopped.
+ * <p/>
+ * This assumes that the "outcome" is not {@code null}.
+ *
+ * @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 BiConsumer<OperationOutcome, Throwable> callbackStarted(CallbackManager callbacks) {
+
+ return (outcome, thrown) -> {
+
+ if (callbacks.canStart()) {
+ // haven't invoked "start" callback yet
+ outcome.setStart(callbacks.getStartTime());
+ outcome.setEnd(null);
+ params.callbackStarted(outcome);
+ }
+ };
+ }
+
+ /**
+ * Sets the end time of the operation and invokes the callback to indicate that the
+ * operation has completed. Does nothing if the pipeline has been stopped.
+ * <p/>
+ * This assumes that the "outcome" is not {@code null}.
+ * <p/>
+ * Note: the start time must be a reference rather than a plain value, because it's
+ * value must be gotten on-demand, when the returned function is executed at a later
+ * time.
+ *
+ * @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 BiConsumer<OperationOutcome, Throwable> callbackCompleted(CallbackManager callbacks) {
+
+ return (outcome, thrown) -> {
+
+ if (callbacks.canEnd()) {
+ outcome.setStart(callbacks.getStartTime());
+ outcome.setEnd(callbacks.getEndTime());
+ params.callbackCompleted(outcome);
+ }
+ };
+ }
+
+ /**
+ * Sets an operation's outcome and message, based on a throwable.
+ *
+ * @param operation operation to be updated
+ * @return the updated operation
+ */
+ protected OperationOutcome setOutcome(OperationOutcome operation, Throwable thrown) {
+ PolicyResult result = (isTimeout(thrown) ? PolicyResult.FAILURE_TIMEOUT : PolicyResult.FAILURE_EXCEPTION);
+ return setOutcome(operation, result);
+ }
+
+ /**
+ * Sets an operation's outcome and default message based on the result.
+ *
+ * @param operation operation to be updated
+ * @param result result of the operation
+ * @return the updated operation
+ */
+ public OperationOutcome setOutcome(OperationOutcome operation, PolicyResult result) {
+ logger.trace("{}: set outcome {} for {}", getFullName(), result, params.getRequestId());
+ operation.setResult(result);
+ operation.setMessage(result == PolicyResult.SUCCESS ? ControlLoopOperation.SUCCESS_MSG
+ : ControlLoopOperation.FAILED_MSG);
+
+ return operation;
+ }
+
+ /**
+ * Determines if a throwable is due to a timeout.
+ *
+ * @param thrown throwable of interest
+ * @return {@code true} if the throwable is due to a timeout, {@code false} otherwise
+ */
+ protected boolean isTimeout(Throwable thrown) {
+ if (thrown instanceof CompletionException) {
+ thrown = thrown.getCause();
+ }
+
+ return (thrown instanceof TimeoutException);
+ }
+
+ // these may be overridden by subclasses or junit tests
+
+ /**
+ * Gets the retry count.
+ *
+ * @param retry retry, extracted from the parameters, or {@code null}
+ * @return the number of retries, or {@code 0} if no retries were specified
+ */
+ protected int getRetry(Integer retry) {
+ return (retry == null ? 0 : retry);
+ }
+
+ /**
+ * Gets the retry wait, in milliseconds.
+ *
+ * @return the retry wait, in milliseconds
+ */
+ protected long getRetryWaitMs() {
+ return DEFAULT_RETRY_WAIT_MS;
+ }
+
+ /**
+ * Gets the operation timeout.
+ *
+ * @param timeoutSec timeout, in seconds, extracted from the parameters, or
+ * {@code null}
+ * @return the operation timeout, in milliseconds, or {@code 0} if no timeout was
+ * specified
+ */
+ protected long getTimeoutMs(Integer timeoutSec) {
+ return (timeoutSec == null ? 0 : TimeUnit.MILLISECONDS.convert(timeoutSec, TimeUnit.SECONDS));
+ }
+}
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 df5258d..3e15c1b 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,57 +20,24 @@
package org.onap.policy.controlloop.actorserviceprovider.impl;
-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.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.PolicyResult;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
- * 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.
+ * Partial implementation of an operator.
*/
public abstract class OperatorPartial extends StartConfigPartial<Map<String, Object>> implements Operator {
- private static final Logger logger = LoggerFactory.getLogger(OperatorPartial.class);
-
/**
* 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.
+ * The "get" method may be overridden by junit tests.
*/
- @Getter(AccessLevel.PROTECTED)
- @Setter(AccessLevel.PROTECTED)
- private Executor blockingExecutor = command -> {
+ @Getter
+ private final Executor blockingExecutor = command -> {
Thread thread = new Thread(command);
thread.setDaemon(true);
thread.start();
@@ -125,721 +92,4 @@
protected void doShutdown() {
// do nothing
}
-
- @Override
- public final CompletableFuture<OperationOutcome> startOperation(ControlLoopOperationParams params) {
- if (!isAlive()) {
- throw new IllegalStateException("operation is not running: " + getFullName());
- }
-
- // allocate a controller for the entire operation
- final PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
-
- CompletableFuture<OperationOutcome> preproc = startPreprocessorAsync(params);
- if (preproc == null) {
- // no preprocessor required - just start the operation
- return startOperationAttempt(params, controller, 1);
- }
-
- /*
- * 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.
- */
- // @formatter:off
- controller.wrap(preproc)
- .exceptionally(fromException(params, "preprocessor of operation"))
- .thenCompose(handlePreprocessorFailure(params, controller))
- .thenCompose(unusedOutcome -> startOperationAttempt(params, controller, 1));
- // @formatter:on
-
- return controller;
- }
-
- /**
- * Handles a failure in the preprocessor pipeline. If a failure occurred, then it
- * 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<OperationOutcome, CompletableFuture<OperationOutcome>> handlePreprocessorFailure(
- ControlLoopOperationParams params, PipelineControllerFuture<OperationOutcome> controller) {
-
- return outcome -> {
-
- if (outcome != null && isSuccess(outcome)) {
- logger.trace("{}: preprocessor succeeded for {}", getFullName(), params.getRequestId());
- return CompletableFuture.completedFuture(outcome);
- }
-
- logger.warn("preprocessor failed, discontinuing operation {} for {}", getFullName(), params.getRequestId());
-
- final Executor executor = params.getExecutor();
- final CallbackManager callbacks = new CallbackManager();
-
- // propagate "stop" to the callbacks
- controller.add(callbacks);
-
- final OperationOutcome outcome2 = params.makeOutcome();
-
- // TODO need a FAILURE_MISSING_DATA (e.g., A&AI)
-
- outcome2.setResult(PolicyResult.FAILURE_GUARD);
- outcome2.setMessage(outcome != null ? outcome.getMessage() : null);
-
- // @formatter:off
- CompletableFuture.completedFuture(outcome2)
- .whenCompleteAsync(callbackStarted(params, callbacks), executor)
- .whenCompleteAsync(callbackCompleted(params, callbacks), executor)
- .whenCompleteAsync(controller.delayedComplete(), executor);
- // @formatter:on
-
- return new CompletableFuture<>();
- };
- }
-
- /**
- * Invokes the operation's preprocessor step(s) as a "future". This method simply
- * returns {@code null}.
- * <p/>
- * This method assumes the following:
- * <ul>
- * <li>the operator is alive</li>
- * <li>exceptions generated within the pipeline will be handled by the invoker</li>
- * </ul>
- *
- * @param params operation parameters
- * @return a function that will start the preprocessor and returns its outcome, or
- * {@code null} if this operation needs no preprocessor
- */
- protected CompletableFuture<OperationOutcome> startPreprocessorAsync(ControlLoopOperationParams params) {
- return null;
- }
-
- /**
- * Starts the operation attempt, with no preprocessor. When all retries complete, it
- * will complete the controller.
- *
- * @param params operation parameters
- * @param controller controller for all operation attempts
- * @param attempt attempt number, typically starting with 1
- * @return a future that will return the final result of all attempts
- */
- private CompletableFuture<OperationOutcome> startOperationAttempt(ControlLoopOperationParams params,
- PipelineControllerFuture<OperationOutcome> controller, int attempt) {
-
- // propagate "stop" to the operation attempt
- controller.wrap(startAttemptWithoutRetries(params, attempt))
- .thenCompose(retryOnFailure(params, controller, attempt));
-
- return controller;
- }
-
- /**
- * Starts the operation attempt, without doing any retries.
- *
- * @param params operation parameters
- * @param attempt attempt number, typically starting with 1
- * @return a future that will return the result of a single operation attempt
- */
- private CompletableFuture<OperationOutcome> startAttemptWithoutRetries(ControlLoopOperationParams params,
- int attempt) {
-
- logger.info("{}: start operation attempt {} for {}", getFullName(), attempt, params.getRequestId());
-
- final Executor executor = params.getExecutor();
- final OperationOutcome outcome = params.makeOutcome();
- final CallbackManager callbacks = new CallbackManager();
-
- // this operation attempt gets its own controller
- final PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
-
- // propagate "stop" to the callbacks
- controller.add(callbacks);
-
- // @formatter:off
- 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.getTimeoutSec());
- if (timeoutMillis > 0) {
- logger.info("{}: set timeout to {}ms for {}", getFullName(), timeoutMillis, params.getRequestId());
- future = future.orTimeout(timeoutMillis, TimeUnit.MILLISECONDS);
- }
-
- /*
- * Note: we re-invoke callbackStarted() just to be sure the callback is invoked
- * before callbackCompleted() is invoked.
- *
- * Note: no need to remove "callbacks" from the pipeline, as we're going to stop
- * the pipeline as the last step anyway.
- */
-
- // @formatter:off
- 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
-
- return controller;
- }
-
- /**
- * Determines if the outcome was successful.
- *
- * @param outcome outcome to examine
- * @return {@code true} if the outcome was successful
- */
- protected boolean isSuccess(OperationOutcome outcome) {
- return (outcome.getResult() == PolicyResult.SUCCESS);
- }
-
- /**
- * Determines if the outcome was a failure for this operator.
- *
- * @param outcome outcome to examine, or {@code null}
- * @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(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)} 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>
- * <li>the operator is alive</li>
- * <li>verifyRunning() has been invoked</li>
- * <li>callbackStarted() has been invoked</li>
- * <li>the invoker will perform appropriate timeout checks</li>
- * <li>exceptions generated within the pipeline will be handled by the invoker</li>
- * </ul>
- *
- * @param params operation parameters
- * @param attempt attempt number, typically starting with 1
- * @return a function that will start the operation and return its result when
- * complete
- */
- protected CompletableFuture<OperationOutcome> startOperationAsync(ControlLoopOperationParams params, int attempt,
- OperationOutcome outcome) {
-
- 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
- * 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 OperationOutcome doOperation(ControlLoopOperationParams params, int attempt, OperationOutcome operation) {
-
- throw new UnsupportedOperationException("start operation " + getFullName());
- }
-
- /**
- * Sets the outcome status to FAILURE_RETRIES, if the current operation outcome is
- * FAILURE, assuming the policy specifies retries and the retry count has been
- * exhausted.
- *
- * @param params operation parameters
- * @param attempt latest attempt number, starting with 1
- * @return a function to get the next future to execute
- */
- private Function<OperationOutcome, OperationOutcome> setRetryFlag(ControlLoopOperationParams params, int attempt) {
-
- return operation -> {
- if (operation != null && !isActorFailed(operation)) {
- /*
- * wrong type or wrong operation - just leave it as is. No need to log
- * anything here, as retryOnFailure() will log a message
- */
- return operation;
- }
-
- // get a non-null operation
- OperationOutcome oper2;
- if (operation != null) {
- oper2 = operation;
- } else {
- oper2 = params.makeOutcome();
- oper2.setResult(PolicyResult.FAILURE);
- }
-
- 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.setResult(PolicyResult.FAILURE_RETRIES);
- }
-
- return oper2;
- };
- }
-
- /**
- * Restarts the operation if it was a FAILURE. Assumes that
- * {@link #setRetryFlag(ControlLoopOperationParams, int)} was previously invoked, and
- * thus that the "operation" is not {@code null}.
- *
- * @param params operation parameters
- * @param controller controller for all of the retries
- * @param attempt latest attempt number, starting with 1
- * @return a function to get the next future to execute
- */
- 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());
- controller.complete(operation);
- return new CompletableFuture<>();
- }
-
- 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());
- controller.complete(operation);
- return new CompletableFuture<>();
- }
-
-
- /*
- * Retry the operation.
- */
- logger.info("retry operation {} for {}", getFullName(), params.getRequestId());
-
- return startOperationAttempt(params, controller, attempt + 1);
- };
- }
-
- /**
- * Converts an exception into an operation outcome, returning a copy of the outcome to
- * prevent background jobs from changing it.
- *
- * @param params operation parameters
- * @param type type of item throwing the exception
- * @return a function that will convert an exception into an operation outcome
- */
- private Function<Throwable, OperationOutcome> fromException(ControlLoopOperationParams params, String type) {
-
- 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) {
-
- // 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);
- }
-
- CompletableFuture.allOf(futures2).whenComplete(combineOutcomes(params, controller, outcomes));
-
- return controller;
- }
-
- /**
- * 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 future future to be completed with the combined result
- * @param outcomes outcomes to be examined
- */
- private BiConsumer<Void, Throwable> combineOutcomes(ControlLoopOperationParams params,
- CompletableFuture<OperationOutcome> future, OperationOutcome[] outcomes) {
-
- return (unused, thrown) -> {
- if (thrown != null) {
- future.completeExceptionally(thrown);
- return;
- }
-
- // identify the outcome with the highest priority
- OperationOutcome outcome = outcomes[0];
- int priority = detmPriority(outcome);
-
- // 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);
-
- 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);
- };
- }
-
- /**
- * Determines the priority of an outcome based on its result.
- *
- * @param outcome outcome to examine, or {@code null}
- * @return the outcome's priority
- */
- protected int detmPriority(OperationOutcome outcome) {
- if (outcome == null) {
- return 1;
- }
-
- 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 complete before canceling so that cancel() doesn't cause controller to
- * complete
- */
- controller.complete(outcome);
- task.cancel(false);
- return new CompletableFuture<>();
- }
-
- return controller.wrap(task);
- }
-
- /**
- * 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 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 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<>();
- }
-
- if (checkSuccess && !isSuccess(outcome)) {
- controller.complete(outcome);
- return new CompletableFuture<>();
- }
-
- return controller.wrap(task.apply(outcome));
- };
- }
-
- /**
- * Sets the start time of the operation and invokes the callback to indicate that the
- * operation has started. Does nothing if the pipeline has been stopped.
- * <p/>
- * This assumes that the "outcome" is not {@code null}.
- *
- * @param params operation parameters
- * @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 BiConsumer<OperationOutcome, Throwable> callbackStarted(ControlLoopOperationParams params,
- CallbackManager callbacks) {
-
- return (outcome, thrown) -> {
-
- if (callbacks.canStart()) {
- // haven't invoked "start" callback yet
- outcome.setStart(callbacks.getStartTime());
- outcome.setEnd(null);
- params.callbackStarted(outcome);
- }
- };
- }
-
- /**
- * Sets the end time of the operation and invokes the callback to indicate that the
- * operation has completed. Does nothing if the pipeline has been stopped.
- * <p/>
- * This assumes that the "outcome" is not {@code null}.
- * <p/>
- * Note: the start time must be a reference rather than a plain value, because it's
- * value must be gotten on-demand, when the returned function is executed at a later
- * time.
- *
- * @param params operation parameters
- * @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 BiConsumer<OperationOutcome, Throwable> callbackCompleted(ControlLoopOperationParams params,
- CallbackManager callbacks) {
-
- return (outcome, thrown) -> {
-
- if (callbacks.canEnd()) {
- outcome.setStart(callbacks.getStartTime());
- outcome.setEnd(callbacks.getEndTime());
- params.callbackCompleted(outcome);
- }
- };
- }
-
- /**
- * Sets an operation's outcome and message, based on a throwable.
- *
- * @param params operation parameters
- * @param operation operation to be updated
- * @return the updated 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);
- }
-
- /**
- * Sets an operation's outcome and default message based on the result.
- *
- * @param params operation parameters
- * @param operation operation to be updated
- * @param result result of the operation
- * @return the updated operation
- */
- protected OperationOutcome setOutcome(ControlLoopOperationParams params, OperationOutcome operation,
- PolicyResult result) {
- logger.trace("{}: set outcome {} for {}", getFullName(), result, params.getRequestId());
- operation.setResult(result);
- operation.setMessage(result == PolicyResult.SUCCESS ? ControlLoopOperation.SUCCESS_MSG
- : ControlLoopOperation.FAILED_MSG);
-
- return operation;
- }
-
- /**
- * Determines if a throwable is due to a timeout.
- *
- * @param thrown throwable of interest
- * @return {@code true} if the throwable is due to a timeout, {@code false} otherwise
- */
- protected boolean isTimeout(Throwable thrown) {
- if (thrown instanceof CompletionException) {
- thrown = thrown.getCause();
- }
-
- return (thrown instanceof TimeoutException);
- }
-
- // these may be overridden by junit tests
-
- /**
- * Gets the operation timeout. Subclasses may override this method to obtain the
- * timeout in some other way (e.g., through configuration properties).
- *
- * @param timeoutSec timeout, in seconds, or {@code null}
- * @return the operation timeout, in milliseconds
- */
- protected long getTimeOutMillis(Integer timeoutSec) {
- return (timeoutSec == null ? 0 : TimeUnit.MILLISECONDS.convert(timeoutSec, TimeUnit.SECONDS));
- }
}
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 57fce40..9259160 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
@@ -148,7 +148,8 @@
return actorService
.getActor(getActor())
.getOperator(getOperation())
- .startOperation(this);
+ .buildOperation(this)
+ .start();
// @formatter:on
}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpActorParams.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpActorParams.java
index da4fb4f..275c8bc 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpActorParams.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpActorParams.java
@@ -51,7 +51,7 @@
* indicates that it should wait forever. The default is zero.
*/
@Min(0)
- private long timeoutSec = 0;
+ private int timeoutSec = 0;
/**
* Maps the operation name to its URI path.
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpParams.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpParams.java
index 695ffe4..93711c0 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpParams.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpParams.java
@@ -53,7 +53,7 @@
*/
@Min(0)
@Builder.Default
- private long timeoutSec = 0;
+ private int timeoutSec = 0;
/**
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/spi/Actor.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/spi/Actor.java
index 620950a..53bee5f 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/spi/Actor.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/spi/Actor.java
@@ -22,7 +22,6 @@
package org.onap.policy.controlloop.actorserviceprovider.spi;
import java.util.Collection;
-
import java.util.List;
import java.util.Map;
import java.util.Set;
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
deleted file mode 100644
index 31c6d20..0000000
--- a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/AsyncResponseHandlerTest.java
+++ /dev/null
@@ -1,172 +0,0 @@
-/*-
- * ============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/UtilTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/UtilTest.java
index 4a3f321..0a2a5a9 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
@@ -39,16 +39,10 @@
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";
/**
@@ -89,82 +83,6 @@
}
@Test
- 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();
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 0d917ad..b462043 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
@@ -24,13 +24,21 @@
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.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
import java.util.Map;
import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
import org.junit.Before;
import org.junit.Test;
import org.onap.policy.controlloop.VirtualControlLoopEvent;
+import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
public class ControlLoopEventContextTest {
private static final UUID REQ_ID = UUID.randomUUID();
@@ -84,4 +92,39 @@
int intValue = context.getProperty("def");
assertEquals(100, intValue);
}
+
+ @Test
+ public void testObtain() {
+ final ControlLoopOperationParams params = mock(ControlLoopOperationParams.class);
+
+ // property is already loaded
+ context.setProperty("obtain-A", "value-A");
+ assertNull(context.obtain("obtain-A", params));
+
+ // new property - should retrieve
+ CompletableFuture<OperationOutcome> future = new CompletableFuture<>();
+ when(params.start()).thenReturn(future);
+ assertSame(future, context.obtain("obtain-B", params));
+
+ // repeat - should get the same future, without invoking start() again
+ assertSame(future, context.obtain("obtain-B", params));
+ verify(params).start();
+
+ // arrange for another invoker to start while this one is starting
+ CompletableFuture<OperationOutcome> future2 = new CompletableFuture<>();
+
+ when(params.start()).thenAnswer(args -> {
+
+ ControlLoopOperationParams params2 = mock(ControlLoopOperationParams.class);
+ when(params2.start()).thenReturn(future2);
+
+ assertSame(future2, context.obtain("obtain-C", params2));
+ return future;
+ });
+
+ assertSame(future2, context.obtain("obtain-C", params));
+
+ // should have canceled the interrupted future
+ assertTrue(future.isCancelled());
+ }
}
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 a209fb0..92cbbe7 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
@@ -42,7 +42,9 @@
import org.junit.Test;
import org.onap.policy.common.parameters.ObjectValidationResult;
import org.onap.policy.common.parameters.ValidationStatus;
+import org.onap.policy.controlloop.actorserviceprovider.Operation;
import org.onap.policy.controlloop.actorserviceprovider.Operator;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
import org.onap.policy.controlloop.actorserviceprovider.parameters.ParameterValidationRuntimeException;
public class ActorImplTest {
@@ -375,10 +377,15 @@
return actor;
}
- private static class MyOper extends OperatorPartial implements Operator {
+ private static class MyOper extends OperatorPartial {
public MyOper(String name) {
super(ACTOR_NAME, name);
}
+
+ @Override
+ public Operation buildOperation(ControlLoopOperationParams params) {
+ return null;
+ }
}
}
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
index 2da7899..8ce3b32 100644
--- 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
@@ -38,7 +38,7 @@
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 static final int TIMEOUT = 10;
private HttpActor actor;
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperationTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperationTest.java
new file mode 100644
index 0000000..19f781d
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperationTest.java
@@ -0,0 +1,781 @@
+/*-
+ * ============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.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+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 static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+import ch.qos.logback.classic.Logger;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.UUID;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.DELETE;
+import javax.ws.rs.GET;
+import javax.ws.rs.POST;
+import javax.ws.rs.PUT;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.client.Entity;
+import javax.ws.rs.client.InvocationCallback;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.core.Response.Status;
+import lombok.Getter;
+import lombok.Setter;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+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.Coder;
+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.common.utils.test.log.logback.ExtractAppender;
+import org.onap.policy.controlloop.VirtualControlLoopEvent;
+import org.onap.policy.controlloop.actorserviceprovider.Operation;
+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.slf4j.LoggerFactory;
+
+public class HttpOperationTest {
+
+ private static final IllegalStateException EXPECTED_EXCEPTION = new IllegalStateException("expected exception");
+ private static final String ACTOR = "my-actor";
+ private static final String OPERATION = "my-name";
+ private static final String HTTP_CLIENT = "my-client";
+ private static final String HTTP_NO_SERVER = "my-http-no-server-client";
+ private static final String MEDIA_TYPE_APPLICATION_JSON = "application/json";
+ private static final String MY_REQUEST = "my-request";
+ private static final String BASE_URI = "oper";
+ private static final String PATH = "/my-path";
+ private static final String TEXT = "my-text";
+ private static final UUID REQ_ID = UUID.randomUUID();
+
+ /**
+ * Used to attach an appender to the class' logger.
+ */
+ private static final Logger logger = (Logger) LoggerFactory.getLogger(HttpOperation.class);
+ private static final ExtractAppender appender = new ExtractAppender();
+
+ /**
+ * {@code True} if the server should reject the request, {@code false} otherwise.
+ */
+ private static boolean rejectRequest;
+
+ // call counts of each method type in the server
+ private static int nget;
+ private static int npost;
+ private static int nput;
+ private static int ndelete;
+
+ @Mock
+ private HttpClient client;
+
+ @Mock
+ private Response response;
+
+ private VirtualControlLoopEvent event;
+ private ControlLoopEventContext context;
+ private ControlLoopOperationParams params;
+ private OperationOutcome outcome;
+ private AtomicReference<InvocationCallback<Response>> callback;
+ private Future<Response> future;
+ private HttpOperator operator;
+ private MyGetOperation<String> oper;
+
+ /**
+ * Starts the 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);
+
+ if (!NetworkUtil.isTcpPortOpen("localhost", port, 100, 100)) {
+ HttpServletServerFactoryInstance.getServerFactory().destroy();
+ throw new IllegalStateException("server is not running");
+ }
+
+ /*
+ * Start the clients, one to the server, and one to a non-existent server.
+ */
+ TopicParamsBuilder builder = BusTopicParams.builder().managed(true).hostname("localhost").basePath(BASE_URI)
+ .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());
+
+ /**
+ * Attach appender to the logger.
+ */
+ appender.setContext(logger.getLoggerContext());
+ appender.start();
+
+ logger.addAppender(appender);
+ }
+
+ /**
+ * Destroys the Http factories and stops the appender.
+ */
+ @AfterClass
+ public static void tearDownAfterClass() {
+ appender.stop();
+
+ HttpClientFactoryInstance.getClientFactory().destroy();
+ HttpServletServerFactoryInstance.getServerFactory().destroy();
+ }
+
+ /**
+ * Initializes fields, including {@link #oper}, and resets the static fields used by
+ * the REST server.
+ */
+ @Before
+ public void setUp() {
+ MockitoAnnotations.initMocks(this);
+
+ appender.clearExtractions();
+
+ rejectRequest = false;
+ nget = 0;
+ npost = 0;
+ nput = 0;
+ ndelete = 0;
+
+ when(response.readEntity(String.class)).thenReturn(TEXT);
+ when(response.getStatus()).thenReturn(200);
+
+ event = new VirtualControlLoopEvent();
+ event.setRequestId(REQ_ID);
+
+ context = new ControlLoopEventContext(event);
+ params = ControlLoopOperationParams.builder().actor(ACTOR).operation(OPERATION).context(context).build();
+
+ outcome = params.makeOutcome();
+
+ callback = new AtomicReference<>();
+ future = new CompletableFuture<>();
+
+ operator = new HttpOperator(ACTOR, OPERATION) {
+ @Override
+ public Operation buildOperation(ControlLoopOperationParams params) {
+ return null;
+ }
+
+ @Override
+ public HttpClient getClient() {
+ return client;
+ }
+ };
+
+ initOper(operator, HTTP_CLIENT);
+
+ oper = new MyGetOperation<>(String.class);
+ }
+
+ @Test
+ public void testHttpOperator() {
+ assertEquals(ACTOR, oper.getActorName());
+ assertEquals(OPERATION, oper.getName());
+ assertEquals(ACTOR + "." + OPERATION, oper.getFullName());
+ }
+
+ @Test
+ public void testMakeHeaders() {
+ assertEquals(Collections.emptyMap(), oper.makeHeaders());
+ }
+
+ @Test
+ public void testMakePath() {
+ assertEquals(PATH, oper.makePath());
+ }
+
+ @Test
+ public void testMakeUrl() {
+ // use a real client
+ client = HttpClientFactoryInstance.getClientFactory().get(HTTP_CLIENT);
+
+ assertThat(oper.makeUrl()).endsWith("/" + BASE_URI + PATH);
+ }
+
+ @Test
+ public void testDoConfigureMapOfStringObject_testGetClient_testGetPath_testGetTimeoutMs() {
+
+ // no default yet
+ assertEquals(0L, oper.getTimeoutMs(null));
+ assertEquals(0L, oper.getTimeoutMs(0));
+
+ // should use given value
+ assertEquals(20 * 1000L, oper.getTimeoutMs(20));
+
+ // indicate we have a timeout value
+ operator = spy(operator);
+ when(operator.getTimeoutMs()).thenReturn(30L);
+
+ oper = new MyGetOperation<String>(String.class);
+
+ // should use default
+ assertEquals(30L, oper.getTimeoutMs(null));
+ assertEquals(30L, oper.getTimeoutMs(0));
+
+ // should use given value
+ assertEquals(40 * 1000L, oper.getTimeoutMs(40));
+ }
+
+ /**
+ * Tests handleResponse() when it completes.
+ */
+ @Test
+ public void testHandleResponseComplete() throws Exception {
+ CompletableFuture<OperationOutcome> future2 = oper.handleResponse(outcome, PATH, cb -> {
+ callback.set(cb);
+ return future;
+ });
+
+ assertFalse(future2.isDone());
+ assertNotNull(callback.get());
+ callback.get().completed(response);
+
+ assertSame(outcome, future2.get(5, TimeUnit.SECONDS));
+
+ assertEquals(PolicyResult.SUCCESS, outcome.getResult());
+ }
+
+ /**
+ * Tests handleResponse() when it fails.
+ */
+ @Test
+ public void testHandleResponseFailed() throws Exception {
+ CompletableFuture<OperationOutcome> future2 = oper.handleResponse(outcome, PATH, cb -> {
+ callback.set(cb);
+ return future;
+ });
+
+ assertFalse(future2.isDone());
+ assertNotNull(callback.get());
+ callback.get().failed(EXPECTED_EXCEPTION);
+
+ assertThatThrownBy(() -> future2.get(5, TimeUnit.SECONDS)).hasCause(EXPECTED_EXCEPTION);
+
+ // future and future2 may be completed in parallel so we must wait again
+ assertThatThrownBy(() -> future.get(5, TimeUnit.SECONDS)).isInstanceOf(CancellationException.class);
+ assertTrue(future.isCancelled());
+ }
+
+ /**
+ * Tests processResponse() when it's a success and the response type is a String.
+ */
+ @Test
+ public void testProcessResponseSuccessString() {
+ assertSame(outcome, oper.processResponse(outcome, PATH, response));
+ assertEquals(PolicyResult.SUCCESS, outcome.getResult());
+ }
+
+ /**
+ * Tests processResponse() when it's a failure.
+ */
+ @Test
+ public void testProcessResponseFailure() {
+ when(response.getStatus()).thenReturn(555);
+ assertSame(outcome, oper.processResponse(outcome, PATH, response));
+ assertEquals(PolicyResult.FAILURE, outcome.getResult());
+ }
+
+ /**
+ * Tests processResponse() when the decoder succeeds.
+ */
+ @Test
+ public void testProcessResponseDecodeOk() throws CoderException {
+ when(response.readEntity(String.class)).thenReturn("10");
+
+ MyGetOperation<Integer> oper2 = new MyGetOperation<>(Integer.class);
+
+ assertSame(outcome, oper2.processResponse(outcome, PATH, response));
+ assertEquals(PolicyResult.SUCCESS, outcome.getResult());
+ }
+
+ /**
+ * Tests processResponse() when the decoder throws an exception.
+ */
+ @Test
+ public void testProcessResponseDecodeExcept() throws CoderException {
+ MyGetOperation<Integer> oper2 = new MyGetOperation<>(Integer.class);
+
+ assertSame(outcome, oper2.processResponse(outcome, PATH, response));
+ assertEquals(PolicyResult.FAILURE_EXCEPTION, outcome.getResult());
+ }
+
+ @Test
+ public void testPostProcessResponse() {
+ assertThatCode(() -> oper.postProcessResponse(outcome, PATH, null, null)).doesNotThrowAnyException();
+ }
+
+ @Test
+ public void testIsSuccess() {
+ when(response.getStatus()).thenReturn(200);
+ assertTrue(oper.isSuccess(response, null));
+
+ when(response.getStatus()).thenReturn(555);
+ assertFalse(oper.isSuccess(response, null));
+ }
+
+ /**
+ * Tests a GET.
+ */
+ @Test
+ public void testGet() throws Exception {
+ // use a real client
+ client = HttpClientFactoryInstance.getClientFactory().get(HTTP_CLIENT);
+
+ MyGetOperation<MyResponse> oper2 = new MyGetOperation<>(MyResponse.class);
+
+ OperationOutcome outcome = runOperation(oper2);
+ assertNotNull(outcome);
+ assertEquals(1, nget);
+ assertEquals(PolicyResult.SUCCESS, outcome.getResult());
+ }
+
+ /**
+ * Tests a DELETE.
+ */
+ @Test
+ public void testDelete() throws Exception {
+ // use a real client
+ client = HttpClientFactoryInstance.getClientFactory().get(HTTP_CLIENT);
+
+ MyDeleteOperation oper2 = new MyDeleteOperation();
+
+ OperationOutcome outcome = runOperation(oper2);
+ assertNotNull(outcome);
+ assertEquals(1, ndelete);
+ assertEquals(PolicyResult.SUCCESS, outcome.getResult());
+ }
+
+ /**
+ * Tests a POST.
+ */
+ @Test
+ public void testPost() throws Exception {
+ // use a real client
+ client = HttpClientFactoryInstance.getClientFactory().get(HTTP_CLIENT);
+
+ MyPostOperation oper2 = new MyPostOperation();
+
+ OperationOutcome outcome = runOperation(oper2);
+ assertNotNull(outcome);
+ assertEquals(1, npost);
+ assertEquals(PolicyResult.SUCCESS, outcome.getResult());
+ }
+
+ /**
+ * Tests a PUT.
+ */
+ @Test
+ public void testPut() throws Exception {
+ // use a real client
+ client = HttpClientFactoryInstance.getClientFactory().get(HTTP_CLIENT);
+
+ MyPutOperation oper2 = new MyPutOperation();
+
+ OperationOutcome outcome = runOperation(oper2);
+ assertNotNull(outcome);
+ assertEquals(1, nput);
+ assertEquals(PolicyResult.SUCCESS, outcome.getResult());
+ }
+
+ @Test
+ public void testLogRestRequest() throws CoderException {
+ // log structured data
+ appender.clearExtractions();
+ oper.logRestRequest(PATH, new MyRequest());
+ List<String> output = appender.getExtracted();
+ assertEquals(1, output.size());
+
+ assertThat(output.get(0)).contains(PATH).contains("{\n \"input\": \"some input\"\n}");
+
+ // log a plain string
+ appender.clearExtractions();
+ oper.logRestRequest(PATH, MY_REQUEST);
+ output = appender.getExtracted();
+ assertEquals(1, output.size());
+
+ assertThat(output.get(0)).contains(PATH).contains(MY_REQUEST);
+
+ // log a null request
+ appender.clearExtractions();
+ oper.logRestRequest(PATH, null);
+ output = appender.getExtracted();
+ assertEquals(1, output.size());
+
+ // exception from coder
+ oper = new MyGetOperation<>(String.class) {
+ @Override
+ protected Coder makeCoder() {
+ return new StandardCoder() {
+ @Override
+ public String encode(Object object, boolean pretty) throws CoderException {
+ throw new CoderException(EXPECTED_EXCEPTION);
+ }
+ };
+ }
+ };
+
+ appender.clearExtractions();
+ oper.logRestRequest(PATH, new MyRequest());
+ output = appender.getExtracted();
+ assertEquals(2, output.size());
+ assertThat(output.get(0)).contains("cannot pretty-print request");
+ assertThat(output.get(1)).contains(PATH);
+ }
+
+ @Test
+ public void testLogRestResponse() throws CoderException {
+ // log structured data
+ appender.clearExtractions();
+ oper.logRestResponse(PATH, new MyResponse());
+ List<String> output = appender.getExtracted();
+ assertEquals(1, output.size());
+
+ assertThat(output.get(0)).contains(PATH).contains("{\n \"output\": \"some output\"\n}");
+
+ // log a plain string
+ appender.clearExtractions();
+ oper.logRestResponse(PATH, MY_REQUEST);
+ output = appender.getExtracted();
+ assertEquals(1, output.size());
+
+ // log a null response
+ appender.clearExtractions();
+ oper.logRestResponse(PATH, null);
+ output = appender.getExtracted();
+ assertEquals(1, output.size());
+
+ assertThat(output.get(0)).contains(PATH).contains("null");
+
+ // exception from coder
+ oper = new MyGetOperation<>(String.class) {
+ @Override
+ protected Coder makeCoder() {
+ return new StandardCoder() {
+ @Override
+ public String encode(Object object, boolean pretty) throws CoderException {
+ throw new CoderException(EXPECTED_EXCEPTION);
+ }
+ };
+ }
+ };
+
+ appender.clearExtractions();
+ oper.logRestResponse(PATH, new MyResponse());
+ output = appender.getExtracted();
+ assertEquals(2, output.size());
+ assertThat(output.get(0)).contains("cannot pretty-print response");
+ assertThat(output.get(1)).contains(PATH);
+ }
+
+ @Test
+ public void testMakeDecoder() {
+ assertNotNull(oper.makeCoder());
+ }
+
+ /**
+ * Gets server properties.
+ *
+ * @param name server name
+ * @param port server port
+ * @return server properties
+ */
+ 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 the given operator.
+ *
+ * @param operator operator to be initialized
+ * @param clientName name of the client which it should use
+ */
+ private void initOper(HttpOperator operator, String clientName) {
+ operator.stop();
+
+ HttpParams params = HttpParams.builder().clientName(clientName).path(PATH).build();
+ Map<String, Object> mapParams = Util.translateToMap(OPERATION, params);
+ operator.configure(mapParams);
+ operator.start();
+ }
+
+ /**
+ * Runs the operation.
+ *
+ * @param operator operator on which to start the operation
+ * @return the outcome of the operation, or {@code null} if it does not complete in
+ * time
+ */
+ private <T> OperationOutcome runOperation(HttpOperation<T> operator)
+ throws InterruptedException, ExecutionException, TimeoutException {
+
+ CompletableFuture<OperationOutcome> future = operator.start();
+
+ return future.get(5, TimeUnit.SECONDS);
+ }
+
+ @Getter
+ @Setter
+ public static class MyRequest {
+ private String input = "some input";
+ }
+
+ @Getter
+ @Setter
+ public static class MyResponse {
+ private String output = "some output";
+ }
+
+ private class MyGetOperation<T> extends HttpOperation<T> {
+ public MyGetOperation(Class<T> responseClass) {
+ super(HttpOperationTest.this.params, HttpOperationTest.this.operator, responseClass);
+ }
+
+ @Override
+ protected CompletableFuture<OperationOutcome> startOperationAsync(int attempt, OperationOutcome outcome) {
+ Map<String, Object> headers = makeHeaders();
+
+ headers.put("Accept", MediaType.APPLICATION_JSON);
+ String url = makeUrl();
+
+ logRestRequest(url, null);
+
+ // @formatter:off
+ return handleResponse(outcome, url,
+ callback -> operator.getClient().get(callback, makePath(), headers));
+ // @formatter:on
+ }
+ }
+
+ private class MyPostOperation extends HttpOperation<MyResponse> {
+ public MyPostOperation() {
+ super(HttpOperationTest.this.params, HttpOperationTest.this.operator, MyResponse.class);
+ }
+
+ @Override
+ protected CompletableFuture<OperationOutcome> startOperationAsync(int attempt, OperationOutcome outcome) {
+
+ MyRequest request = new MyRequest();
+
+ Entity<MyRequest> entity = Entity.entity(request, MediaType.APPLICATION_JSON);
+
+ Map<String, Object> headers = makeHeaders();
+
+ headers.put("Accept", MediaType.APPLICATION_JSON);
+ String url = makeUrl();
+
+ logRestRequest(url, request);
+
+ // @formatter:off
+ return handleResponse(outcome, url,
+ callback -> operator.getClient().post(callback, makePath(), entity, headers));
+ // @formatter:on
+ }
+ }
+
+ private class MyPutOperation extends HttpOperation<MyResponse> {
+ public MyPutOperation() {
+ super(HttpOperationTest.this.params, HttpOperationTest.this.operator, MyResponse.class);
+ }
+
+ @Override
+ protected CompletableFuture<OperationOutcome> startOperationAsync(int attempt, OperationOutcome outcome) {
+
+ MyRequest request = new MyRequest();
+
+ Entity<MyRequest> entity = Entity.entity(request, MediaType.APPLICATION_JSON);
+
+ Map<String, Object> headers = makeHeaders();
+
+ headers.put("Accept", MediaType.APPLICATION_JSON);
+ String url = makeUrl();
+
+ logRestRequest(url, request);
+
+ // @formatter:off
+ return handleResponse(outcome, url,
+ callback -> operator.getClient().put(callback, makePath(), entity, headers));
+ // @formatter:on
+ }
+ }
+
+ private class MyDeleteOperation extends HttpOperation<String> {
+ public MyDeleteOperation() {
+ super(HttpOperationTest.this.params, HttpOperationTest.this.operator, String.class);
+ }
+
+ @Override
+ protected CompletableFuture<OperationOutcome> startOperationAsync(int attempt, OperationOutcome outcome) {
+ Map<String, Object> headers = makeHeaders();
+
+ headers.put("Accept", MediaType.APPLICATION_JSON);
+ String url = makeUrl();
+
+ logRestRequest(url, null);
+
+ // @formatter:off
+ return handleResponse(outcome, url,
+ callback -> operator.getClient().delete(callback, makePath(), headers));
+ // @formatter:on
+ }
+ }
+
+ /**
+ * Simulator.
+ */
+ @Path("/" + BASE_URI)
+ @Produces(MEDIA_TYPE_APPLICATION_JSON)
+ @Consumes(value = {MEDIA_TYPE_APPLICATION_JSON})
+ public static class Server {
+
+ /**
+ * Generates a response to a GET.
+ *
+ * @return resulting response
+ */
+ @GET
+ @Path(PATH)
+ public Response getRequest() {
+ ++nget;
+
+ if (rejectRequest) {
+ return Response.status(Status.BAD_REQUEST).build();
+
+ } else {
+ return Response.status(Status.OK).entity(new MyResponse()).build();
+ }
+ }
+
+ /**
+ * Generates a response to a POST.
+ *
+ * @param request incoming request
+ * @return resulting response
+ */
+ @POST
+ @Path(PATH)
+ public Response postRequest(MyRequest request) {
+ ++npost;
+
+ if (rejectRequest) {
+ return Response.status(Status.BAD_REQUEST).build();
+
+ } else {
+ return Response.status(Status.OK).entity(new MyResponse()).build();
+ }
+ }
+
+ /**
+ * Generates a response to a PUT.
+ *
+ * @param request incoming request
+ * @return resulting response
+ */
+ @PUT
+ @Path(PATH)
+ public Response putRequest(MyRequest request) {
+ ++nput;
+
+ if (rejectRequest) {
+ return Response.status(Status.BAD_REQUEST).build();
+
+ } else {
+ return Response.status(Status.OK).entity(new MyResponse()).build();
+ }
+ }
+
+ /**
+ * Generates a response to a DELETE.
+ *
+ * @return resulting response
+ */
+ @DELETE
+ @Path(PATH)
+ public Response deleteRequest() {
+ ++ndelete;
+
+ if (rejectRequest) {
+ return Response.status(Status.BAD_REQUEST).build();
+
+ } else {
+ return Response.status(Status.OK).entity(new MyResponse()).build();
+ }
+ }
+ }
+}
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
index c006cf3..081bb34 100644
--- 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
@@ -23,19 +23,25 @@
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.assertNotSame;
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 java.util.concurrent.CompletableFuture;
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.VirtualControlLoopEvent;
+import org.onap.policy.controlloop.actorserviceprovider.Operation;
+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.actorserviceprovider.parameters.ParameterValidationRuntimeException;
@@ -43,51 +49,33 @@
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;
+ private static final String HTTP_CLIENT = "my-client";
+ private static final String PATH = "/my-path";
+ private static final int TIMEOUT = 100;
@Mock
private HttpClient client;
- private HttpOperator oper;
+ @Mock
+ private HttpClientFactory factory;
+
+ private MyOperator oper;
/**
- * Initializes fields, including {@link #oper}.
+ * Initializes fields, including {@link #oper}, and resets the static fields used by
+ * the REST server.
*/
@Before
public void setUp() {
MockitoAnnotations.initMocks(this);
- oper = new HttpOperator(ACTOR, OPERATION);
- }
+ when(factory.get(HTTP_CLIENT)).thenReturn(client);
- @Test
- public void testDoConfigureMapOfStringObject_testGetClient_testGetPath_testGetTimeoutSec() {
- assertNull(oper.getClient());
- assertNull(oper.getPath());
- assertEquals(0L, oper.getTimeoutSec());
+ oper = new MyOperator();
- 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();
+ HttpParams params = HttpParams.builder().clientName(HTTP_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
@@ -99,6 +87,78 @@
@Test
public void testGetClient() {
- assertNotNull(oper.getClientFactory());
+ assertNotNull(oper.getClient());
+ }
+
+ @Test
+ public void testMakeOperator() {
+ HttpOperator oper2 = HttpOperator.makeOperator(ACTOR, OPERATION, MyOperation::new);
+ assertNotNull(oper2);
+
+ VirtualControlLoopEvent event = new VirtualControlLoopEvent();
+ ControlLoopEventContext context = new ControlLoopEventContext(event);
+ ControlLoopOperationParams params =
+ ControlLoopOperationParams.builder().actor(ACTOR).operation(OPERATION).context(context).build();
+
+ Operation operation1 = oper2.buildOperation(params);
+ assertNotNull(operation1);
+
+ Operation operation2 = oper2.buildOperation(params);
+ assertNotNull(operation2);
+ assertNotSame(operation1, operation2);
+ }
+
+ @Test
+ public void testDoConfigureMapOfStringObject_testGetClient_testGetPath_testGetTimeoutMs() {
+ // start with an UNCONFIGURED operator
+ oper.shutdown();
+ oper = new MyOperator();
+
+ assertNull(oper.getClient());
+ assertNull(oper.getPath());
+
+ // no timeout yet
+ assertEquals(0L, oper.getTimeoutMs());
+
+ HttpParams params = HttpParams.builder().clientName(HTTP_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());
+
+ // should use given value
+ assertEquals(TIMEOUT * 1000, oper.getTimeoutMs());
+
+ // test invalid parameters
+ paramMap.remove("path");
+ assertThatThrownBy(() -> oper.configure(paramMap)).isInstanceOf(ParameterValidationRuntimeException.class);
+ }
+
+ private class MyOperator extends HttpOperator {
+ public MyOperator() {
+ super(ACTOR, OPERATION);
+ }
+
+ @Override
+ public Operation buildOperation(ControlLoopOperationParams params) {
+ return null;
+ }
+
+ @Override
+ protected HttpClientFactory getClientFactory() {
+ return factory;
+ }
+ }
+
+ private class MyOperation extends HttpOperation<String> {
+ public MyOperation(ControlLoopOperationParams params, HttpOperator operator) {
+ super(params, operator, String.class);
+ }
+
+ @Override
+ protected CompletableFuture<OperationOutcome> startOperationAsync(int attempt, OperationOutcome outcome) {
+ return null;
+ }
}
}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperationPartialTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperationPartialTest.java
new file mode 100644
index 0000000..0d5cb24
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperationPartialTest.java
@@ -0,0 +1,1302 @@
+/*-
+ * ============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.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
+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.assertSame;
+import static org.junit.Assert.assertTrue;
+
+import java.time.Instant;
+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.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+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;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import lombok.Getter;
+import lombok.Setter;
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.policy.common.utils.coder.CoderException;
+import org.onap.policy.common.utils.coder.StandardCoder;
+import org.onap.policy.controlloop.ControlLoopOperation;
+import org.onap.policy.controlloop.VirtualControlLoopEvent;
+import org.onap.policy.controlloop.actorserviceprovider.Operation;
+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.actorserviceprovider.pipeline.PipelineControllerFuture;
+import org.onap.policy.controlloop.policy.PolicyResult;
+
+public class OperationPartialTest {
+ private static final int MAX_PARALLEL_REQUESTS = 10;
+ 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 int TIMEOUT = 1000;
+ private static final UUID REQ_ID = UUID.randomUUID();
+
+ private static final List<PolicyResult> FAILURE_RESULTS = Arrays.asList(PolicyResult.values()).stream()
+ .filter(result -> result != PolicyResult.SUCCESS).collect(Collectors.toList());
+
+ private VirtualControlLoopEvent event;
+ private ControlLoopEventContext context;
+ private MyExec executor;
+ private ControlLoopOperationParams params;
+
+ private MyOper oper;
+
+ private int numStart;
+ private int numEnd;
+
+ private Instant tstart;
+
+ private OperationOutcome opstart;
+ private OperationOutcome opend;
+
+ private OperatorPartial operator;
+
+ /**
+ * Initializes the fields, including {@link #oper}.
+ */
+ @Before
+ public void setUp() {
+ event = new VirtualControlLoopEvent();
+ event.setRequestId(REQ_ID);
+
+ context = new ControlLoopEventContext(event);
+ executor = new MyExec();
+
+ params = ControlLoopOperationParams.builder().completeCallback(this::completer).context(context)
+ .executor(executor).actor(ACTOR).operation(OPERATION).timeoutSec(TIMEOUT)
+ .startCallback(this::starter).targetEntity(TARGET).build();
+
+ operator = new OperatorPartial(ACTOR, OPERATION) {
+ @Override
+ public Executor getBlockingExecutor() {
+ return executor;
+ }
+
+ @Override
+ public Operation buildOperation(ControlLoopOperationParams params) {
+ return null;
+ }
+ };
+
+ operator.configure(null);
+ operator.start();
+
+ oper = new MyOper();
+
+ tstart = null;
+
+ opstart = null;
+ opend = null;
+ }
+
+ @Test
+ public void testOperatorPartial_testGetActorName_testGetName() {
+ assertEquals(ACTOR, oper.getActorName());
+ assertEquals(OPERATION, oper.getName());
+ assertEquals(ACTOR + "." + OPERATION, oper.getFullName());
+ }
+
+ @Test
+ public void testGetBlockingThread() throws Exception {
+ CompletableFuture<Void> future = new CompletableFuture<>();
+
+ // use the real executor
+ OperatorPartial oper2 = new OperatorPartial(ACTOR, OPERATION) {
+ @Override
+ public Operation buildOperation(ControlLoopOperationParams params) {
+ return null;
+ }
+ };
+
+ oper2.getBlockingExecutor().execute(() -> future.complete(null));
+
+ assertNull(future.get(5, TimeUnit.SECONDS));
+ }
+
+ /**
+ * Exercises the doXxx() methods.
+ */
+ @Test
+ public void testDoXxx() {
+ assertThatCode(() -> operator.doConfigure(null)).doesNotThrowAnyException();
+ assertThatCode(() -> operator.doStart()).doesNotThrowAnyException();
+ assertThatCode(() -> operator.doStop()).doesNotThrowAnyException();
+ assertThatCode(() -> operator.doShutdown()).doesNotThrowAnyException();
+
+ }
+
+ @Test
+ public void testStart() {
+ verifyRun("testStart", 1, 1, PolicyResult.SUCCESS);
+ }
+
+ /**
+ * Tests startOperation() when the operator is not running.
+ */
+ @Test
+ public void testStartNotRunning() {
+ // stop the operator
+ operator.stop();
+
+ assertThatIllegalStateException().isThrownBy(() -> oper.start());
+ }
+
+ /**
+ * Tests startOperation() when the operation has a preprocessor.
+ */
+ @Test
+ public void testStartWithPreprocessor() {
+ AtomicInteger count = new AtomicInteger();
+
+ CompletableFuture<OperationOutcome> preproc = CompletableFuture.supplyAsync(() -> {
+ count.incrementAndGet();
+ return makeSuccess();
+ }, executor);
+
+ oper.setGuard(preproc);
+
+ verifyRun("testStartWithPreprocessor_testStartPreprocessor", 1, 1, PolicyResult.SUCCESS);
+
+ assertEquals(1, count.get());
+ }
+
+ /**
+ * Tests start() with multiple running requests.
+ */
+ @Test
+ public void testStartMultiple() {
+ for (int count = 0; count < MAX_PARALLEL_REQUESTS; ++count) {
+ oper.start();
+ }
+
+ assertTrue(executor.runAll());
+
+ assertNotNull(opstart);
+ assertNotNull(opend);
+ assertEquals(PolicyResult.SUCCESS, opend.getResult());
+
+ assertEquals(MAX_PARALLEL_REQUESTS, numStart);
+ assertEquals(MAX_PARALLEL_REQUESTS, oper.getCount());
+ assertEquals(MAX_PARALLEL_REQUESTS, numEnd);
+ }
+
+ /**
+ * Tests startPreprocessor() when the preprocessor returns a failure.
+ */
+ @Test
+ public void testStartPreprocessorFailure() {
+ oper.setGuard(CompletableFuture.completedFuture(makeFailure()));
+
+ verifyRun("testStartPreprocessorFailure", 1, 0, PolicyResult.FAILURE_GUARD);
+ }
+
+ /**
+ * Tests startPreprocessor() when the preprocessor throws an exception.
+ */
+ @Test
+ public void testStartPreprocessorException() {
+ // arrange for the preprocessor to throw an exception
+ oper.setGuard(CompletableFuture.failedFuture(new IllegalStateException(EXPECTED_EXCEPTION)));
+
+ verifyRun("testStartPreprocessorException", 1, 0, PolicyResult.FAILURE_GUARD);
+ }
+
+ /**
+ * Tests startPreprocessor() when the pipeline is not running.
+ */
+ @Test
+ public void testStartPreprocessorNotRunning() {
+ // arrange for the preprocessor to return success, which will be ignored
+ oper.setGuard(CompletableFuture.completedFuture(makeSuccess()));
+
+ oper.start().cancel(false);
+ assertTrue(executor.runAll());
+
+ assertNull(opstart);
+ assertNull(opend);
+
+ assertEquals(0, numStart);
+ assertEquals(0, oper.getCount());
+ assertEquals(0, numEnd);
+ }
+
+ /**
+ * Tests startPreprocessor() when the preprocessor <b>builder</b> throws an exception.
+ */
+ @Test
+ public void testStartPreprocessorBuilderException() {
+ oper = new MyOper() {
+ @Override
+ protected CompletableFuture<OperationOutcome> startPreprocessorAsync() {
+ throw new IllegalStateException(EXPECTED_EXCEPTION);
+ }
+ };
+
+ assertThatIllegalStateException().isThrownBy(() -> oper.start());
+
+ // should be nothing in the queue
+ assertEquals(0, executor.getQueueLength());
+ }
+
+ @Test
+ public void testStartPreprocessorAsync() {
+ assertNull(oper.startPreprocessorAsync());
+ }
+
+ @Test
+ public void testStartGuardAsync() {
+ assertNull(oper.startGuardAsync());
+ }
+
+ @Test
+ public void testStartOperationAsync() {
+ oper.start();
+ assertTrue(executor.runAll());
+
+ assertEquals(1, oper.getCount());
+ }
+
+ @Test
+ public void testIsSuccess() {
+ OperationOutcome outcome = new OperationOutcome();
+
+ outcome.setResult(PolicyResult.SUCCESS);
+ assertTrue(oper.isSuccess(outcome));
+
+ for (PolicyResult failure : FAILURE_RESULTS) {
+ outcome.setResult(failure);
+ assertFalse("testIsSuccess-" + failure, oper.isSuccess(outcome));
+ }
+ }
+
+ @Test
+ public void testIsActorFailed() {
+ assertFalse(oper.isActorFailed(null));
+
+ OperationOutcome outcome = params.makeOutcome();
+
+ // incorrect outcome
+ outcome.setResult(PolicyResult.SUCCESS);
+ assertFalse(oper.isActorFailed(outcome));
+
+ outcome.setResult(PolicyResult.FAILURE_RETRIES);
+ assertFalse(oper.isActorFailed(outcome));
+
+ // correct outcome
+ outcome.setResult(PolicyResult.FAILURE);
+
+ // incorrect actor
+ outcome.setActor(TARGET);
+ assertFalse(oper.isActorFailed(outcome));
+ outcome.setActor(null);
+ assertFalse(oper.isActorFailed(outcome));
+ outcome.setActor(ACTOR);
+
+ // incorrect operation
+ outcome.setOperation(TARGET);
+ assertFalse(oper.isActorFailed(outcome));
+ outcome.setOperation(null);
+ assertFalse(oper.isActorFailed(outcome));
+ outcome.setOperation(OPERATION);
+
+ // correct values
+ assertTrue(oper.isActorFailed(outcome));
+ }
+
+ @Test
+ public void testDoOperation() {
+ /*
+ * Use an operation that doesn't override doOperation().
+ */
+ OperationPartial oper2 = new OperationPartial(params, operator) {};
+
+ oper2.start();
+ assertTrue(executor.runAll());
+
+ assertNotNull(opend);
+ assertEquals(PolicyResult.FAILURE_EXCEPTION, opend.getResult());
+ }
+
+ @Test
+ public void testTimeout() throws Exception {
+
+ // use a real executor
+ params = params.toBuilder().executor(ForkJoinPool.commonPool()).build();
+
+ // trigger timeout very quickly
+ oper = new MyOper() {
+ @Override
+ protected long getTimeoutMs(Integer timeoutSec) {
+ return 1;
+ }
+
+ @Override
+ protected CompletableFuture<OperationOutcome> startOperationAsync(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;
+ }
+ };
+
+ assertEquals(PolicyResult.FAILURE_TIMEOUT, oper.start().get().getResult());
+ }
+
+ /**
+ * Tests retry functions, when the count is set to zero and retries are exhausted.
+ */
+ @Test
+ public void testSetRetryFlag_testRetryOnFailure_ZeroRetries_testStartOperationAttempt() {
+ params = params.toBuilder().retry(0).build();
+
+ // new params, thus need a new operation
+ oper = new MyOper();
+
+ oper.setMaxFailures(10);
+
+ verifyRun("testSetRetryFlag_testRetryOnFailure_ZeroRetries", 1, 1, PolicyResult.FAILURE);
+ }
+
+ /**
+ * Tests retry functions, when the count is null and retries are exhausted.
+ */
+ @Test
+ public void testSetRetryFlag_testRetryOnFailure_NullRetries() {
+ params = params.toBuilder().retry(null).build();
+
+ // new params, thus need a new operation
+ oper = new MyOper();
+
+ oper.setMaxFailures(10);
+
+ verifyRun("testSetRetryFlag_testRetryOnFailure_NullRetries", 1, 1, PolicyResult.FAILURE);
+ }
+
+ /**
+ * Tests retry functions, when retries are exhausted.
+ */
+ @Test
+ public void testSetRetryFlag_testRetryOnFailure_RetriesExhausted() {
+ final int maxRetries = 3;
+ params = params.toBuilder().retry(maxRetries).build();
+
+ // new params, thus need a new operation
+ oper = new MyOper();
+
+ oper.setMaxFailures(10);
+
+ verifyRun("testSetRetryFlag_testRetryOnFailure_RetriesExhausted", maxRetries + 1, maxRetries + 1,
+ PolicyResult.FAILURE_RETRIES);
+ }
+
+ /**
+ * Tests retry functions, when a success follows some retries.
+ */
+ @Test
+ public void testSetRetryFlag_testRetryOnFailure_SuccessAfterRetries() {
+ params = params.toBuilder().retry(10).build();
+
+ // new params, thus need a new operation
+ oper = new MyOper();
+
+ final int maxFailures = 3;
+ oper.setMaxFailures(maxFailures);
+
+ verifyRun("testSetRetryFlag_testRetryOnFailure_SuccessAfterRetries", maxFailures + 1, maxFailures + 1,
+ PolicyResult.SUCCESS);
+ }
+
+ /**
+ * Tests retry functions, when the outcome is {@code null}.
+ */
+ @Test
+ public void testSetRetryFlag_testRetryOnFailure_NullOutcome() {
+
+ // arrange to return null from doOperation()
+ oper = new MyOper() {
+ @Override
+ protected OperationOutcome doOperation(int attempt, OperationOutcome operation) {
+
+ // update counters
+ super.doOperation(attempt, operation);
+ return null;
+ }
+ };
+
+ verifyRun("testSetRetryFlag_testRetryOnFailure_NullOutcome", 1, 1, PolicyResult.FAILURE, null, noop());
+ }
+
+ @Test
+ public void testSleep() throws Exception {
+ CompletableFuture<Void> future = oper.sleep(-1, TimeUnit.SECONDS);
+ assertTrue(future.isDone());
+ assertNull(future.get());
+
+ // edge case
+ future = oper.sleep(0, TimeUnit.SECONDS);
+ assertTrue(future.isDone());
+ assertNull(future.get());
+
+ /*
+ * Start a second sleep we can use to check the first while it's running.
+ */
+ tstart = Instant.now();
+ future = oper.sleep(100, TimeUnit.MILLISECONDS);
+
+ CompletableFuture<Void> future2 = oper.sleep(10, TimeUnit.MILLISECONDS);
+
+ // wait for second to complete and verify that the first has not completed
+ future2.get();
+ assertFalse(future.isDone());
+
+ // wait for second to complete
+ future.get();
+
+ long diff = Instant.now().toEpochMilli() - tstart.toEpochMilli();
+ assertTrue(diff >= 99);
+ }
+
+ @Test
+ public void testIsSameOperation() {
+ assertFalse(oper.isSameOperation(null));
+
+ OperationOutcome outcome = params.makeOutcome();
+
+ // wrong actor - should be false
+ outcome.setActor(null);
+ assertFalse(oper.isSameOperation(outcome));
+ outcome.setActor(TARGET);
+ assertFalse(oper.isSameOperation(outcome));
+ outcome.setActor(ACTOR);
+
+ // wrong operation - should be null
+ outcome.setOperation(null);
+ assertFalse(oper.isSameOperation(outcome));
+ outcome.setOperation(TARGET);
+ assertFalse(oper.isSameOperation(outcome));
+ outcome.setOperation(OPERATION);
+
+ assertTrue(oper.isSameOperation(outcome));
+ }
+
+ /**
+ * Tests handleFailure() when the outcome is a success.
+ */
+ @Test
+ public void testHandlePreprocessorFailureTrue() {
+ oper.setGuard(CompletableFuture.completedFuture(makeSuccess()));
+ verifyRun("testHandlePreprocessorFailureTrue", 1, 1, PolicyResult.SUCCESS);
+ }
+
+ /**
+ * Tests handleFailure() when the outcome is <i>not</i> a success.
+ */
+ @Test
+ public void testHandlePreprocessorFailureFalse() throws Exception {
+ oper.setGuard(CompletableFuture.completedFuture(makeFailure()));
+ verifyRun("testHandlePreprocessorFailureFalse", 1, 0, PolicyResult.FAILURE_GUARD);
+ }
+
+ /**
+ * Tests handleFailure() when the outcome is {@code null}.
+ */
+ @Test
+ public void testHandlePreprocessorFailureNull() throws Exception {
+ // arrange to return null from the preprocessor
+ oper.setGuard(CompletableFuture.completedFuture(null));
+
+ verifyRun("testHandlePreprocessorFailureNull", 1, 0, PolicyResult.FAILURE_GUARD);
+ }
+
+ @Test
+ public void testFromException() {
+ // arrange to generate an exception when operation runs
+ oper.setGenException(true);
+
+ verifyRun("testFromException", 1, 1, PolicyResult.FAILURE_EXCEPTION);
+ }
+
+ /**
+ * Tests fromException() when there is no exception.
+ */
+ @Test
+ public void testFromExceptionNoExcept() {
+ verifyRun("testFromExceptionNoExcept", 1, 1, PolicyResult.SUCCESS);
+ }
+
+ /**
+ * 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(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(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(tasks);
+ assertTrue(executor.runAll());
+
+ assertTrue(result.isDone());
+ assertSame(outcome, result.get());
+ }
+
+ /**
+ * Tests both flavors of anyOf(), for edge cases: zero items, and one item.
+ */
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testAnyOfEdge() throws Exception {
+ List<CompletableFuture<OperationOutcome>> tasks = new LinkedList<>();
+
+ // zero items: check both using a list and using an array
+ assertThatIllegalArgumentException().isThrownBy(() -> oper.anyOf(tasks));
+ assertThatIllegalArgumentException().isThrownBy(() -> oper.anyOf());
+
+ // one item: : check both using a list and using an array
+ CompletableFuture<OperationOutcome> future1 = new CompletableFuture<>();
+ tasks.add(future1);
+
+ assertSame(future1, oper.anyOf(tasks));
+ assertSame(future1, oper.anyOf(future1));
+ }
+
+ /**
+ * 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(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());
+ }
+
+ /**
+ * Tests both flavors of allOf(), for edge cases: zero items, and one item.
+ */
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testAllOfEdge() throws Exception {
+ List<CompletableFuture<OperationOutcome>> tasks = new LinkedList<>();
+
+ // zero items: check both using a list and using an array
+ assertThatIllegalArgumentException().isThrownBy(() -> oper.allOf(tasks));
+ assertThatIllegalArgumentException().isThrownBy(() -> oper.allOf());
+
+ // one item: : check both using a list and using an array
+ CompletableFuture<OperationOutcome> future1 = new CompletableFuture<>();
+ tasks.add(future1);
+
+ assertSame(future1, oper.allOf(tasks));
+ assertSame(future1, oper.allOf(future1));
+ }
+
+ @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(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(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(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() throws CoderException {
+ 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));
+ }
+
+ /*
+ * Test null result. We can't actually set it to null, because the set() method
+ * won't allow it. Instead, we decode it from a structure.
+ */
+ outcome = new StandardCoder().decode("{\"result\":null}", OperationOutcome.class);
+ assertEquals(1, 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(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(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(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(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(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(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(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 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(controller, false, task).apply(failedOutcome);
+
+ assertTrue(future.isDone());
+ assertSame(taskOutcome, future.get());
+
+ // controller should not be done yet
+ assertFalse(controller.isDone());
+ }
+
+ /**
+ * Tests callbackStarted() when the pipeline has already been stopped.
+ */
+ @Test
+ public void testCallbackStartedNotRunning() {
+ AtomicReference<Future<OperationOutcome>> future = new AtomicReference<>();
+
+ /*
+ * arrange to stop the controller when the start-callback is invoked, but capture
+ * the outcome
+ */
+ params = params.toBuilder().startCallback(oper -> {
+ starter(oper);
+ future.get().cancel(false);
+ }).build();
+
+ // new params, thus need a new operation
+ oper = new MyOper();
+
+ future.set(oper.start());
+ assertTrue(executor.runAll());
+
+ // should have only run once
+ assertEquals(1, numStart);
+ }
+
+ /**
+ * Tests callbackCompleted() when the pipeline has already been stopped.
+ */
+ @Test
+ public void testCallbackCompletedNotRunning() {
+ AtomicReference<Future<OperationOutcome>> future = new AtomicReference<>();
+
+ // arrange to stop the controller when the start-callback is invoked
+ params = params.toBuilder().startCallback(oper -> {
+ future.get().cancel(false);
+ }).build();
+
+ // new params, thus need a new operation
+ oper = new MyOper();
+
+ future.set(oper.start());
+ assertTrue(executor.runAll());
+
+ // should not have been set
+ assertNull(opend);
+ assertEquals(0, numEnd);
+ }
+
+ @Test
+ public void testSetOutcomeControlLoopOperationOutcomeThrowable() {
+ final CompletionException timex = new CompletionException(new TimeoutException(EXPECTED_EXCEPTION));
+
+ OperationOutcome outcome;
+
+ outcome = new OperationOutcome();
+ oper.setOutcome(outcome, timex);
+ assertEquals(ControlLoopOperation.FAILED_MSG, outcome.getMessage());
+ assertEquals(PolicyResult.FAILURE_TIMEOUT, outcome.getResult());
+
+ outcome = new OperationOutcome();
+ oper.setOutcome(outcome, new IllegalStateException(EXPECTED_EXCEPTION));
+ assertEquals(ControlLoopOperation.FAILED_MSG, outcome.getMessage());
+ assertEquals(PolicyResult.FAILURE_EXCEPTION, outcome.getResult());
+ }
+
+ @Test
+ public void testSetOutcomeControlLoopOperationOutcomePolicyResult() {
+ OperationOutcome outcome;
+
+ outcome = new OperationOutcome();
+ oper.setOutcome(outcome, PolicyResult.SUCCESS);
+ assertEquals(ControlLoopOperation.SUCCESS_MSG, outcome.getMessage());
+ assertEquals(PolicyResult.SUCCESS, outcome.getResult());
+
+ for (PolicyResult result : FAILURE_RESULTS) {
+ outcome = new OperationOutcome();
+ oper.setOutcome(outcome, result);
+ assertEquals(result.toString(), ControlLoopOperation.FAILED_MSG, outcome.getMessage());
+ assertEquals(result.toString(), result, outcome.getResult());
+ }
+ }
+
+ @Test
+ public void testIsTimeout() {
+ final TimeoutException timex = new TimeoutException(EXPECTED_EXCEPTION);
+
+ 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)));
+ assertFalse(oper.isTimeout(new CompletionException(new CompletionException(timex))));
+
+ assertTrue(oper.isTimeout(timex));
+ assertTrue(oper.isTimeout(new CompletionException(timex)));
+ }
+
+ @Test
+ public void testGetRetry() {
+ assertEquals(0, oper.getRetry(null));
+ assertEquals(10, oper.getRetry(10));
+ }
+
+ @Test
+ public void testGetRetryWait() {
+ // need an operator that doesn't override the retry time
+ OperationPartial oper2 = new OperationPartial(params, operator) {};
+ assertEquals(OperationPartial.DEFAULT_RETRY_WAIT_MS, oper2.getRetryWaitMs());
+ }
+
+ @Test
+ public void testGetTimeOutMs() {
+ assertEquals(TIMEOUT * 1000, oper.getTimeoutMs(params.getTimeoutSec()));
+
+ params = params.toBuilder().timeoutSec(null).build();
+
+ // new params, thus need a new operation
+ oper = new MyOper();
+
+ assertEquals(0, oper.getTimeoutMs(params.getTimeoutSec()));
+ }
+
+ private void starter(OperationOutcome oper) {
+ ++numStart;
+ tstart = oper.getStart();
+ opstart = oper;
+ }
+
+ private void completer(OperationOutcome oper) {
+ ++numEnd;
+ opend = oper;
+ }
+
+ /**
+ * Gets a function that does nothing.
+ *
+ * @param <T> type of input parameter expected by the function
+ * @return a function that does nothing
+ */
+ private <T> Consumer<T> noop() {
+ return unused -> {
+ };
+ }
+
+ private OperationOutcome makeSuccess() {
+ OperationOutcome outcome = params.makeOutcome();
+ outcome.setResult(PolicyResult.SUCCESS);
+
+ return outcome;
+ }
+
+ private OperationOutcome makeFailure() {
+ OperationOutcome outcome = params.makeOutcome();
+ outcome.setResult(PolicyResult.FAILURE);
+
+ return outcome;
+ }
+
+ /**
+ * 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) {
+
+ String expectedSubRequestId =
+ (expectedResult == PolicyResult.FAILURE_EXCEPTION ? null : String.valueOf(expectedOperations));
+
+ verifyRun(testName, expectedCallbacks, expectedOperations, expectedResult, expectedSubRequestId, noop());
+ }
+
+ /**
+ * Verifies a run.
+ *
+ * @param testName test name
+ * @param expectedCallbacks number of callbacks expected
+ * @param expectedOperations number of operation invocations expected
+ * @param expectedResult expected outcome
+ * @param expectedSubRequestId expected sub request ID
+ * @param manipulator function to modify the future returned by
+ * {@link OperationPartial#start(ControlLoopOperationParams)} before the tasks
+ * in the executor are run
+ */
+ private void verifyRun(String testName, int expectedCallbacks, int expectedOperations, PolicyResult expectedResult,
+ String expectedSubRequestId, Consumer<CompletableFuture<OperationOutcome>> manipulator) {
+
+ CompletableFuture<OperationOutcome> future = oper.start();
+
+ manipulator.accept(future);
+
+ assertTrue(testName, executor.runAll());
+
+ assertEquals(testName, expectedCallbacks, numStart);
+ assertEquals(testName, expectedCallbacks, numEnd);
+
+ if (expectedCallbacks > 0) {
+ assertNotNull(testName, opstart);
+ assertNotNull(testName, opend);
+ assertEquals(testName, expectedResult, opend.getResult());
+
+ assertSame(testName, tstart, opstart.getStart());
+ assertSame(testName, tstart, opend.getStart());
+
+ try {
+ assertTrue(future.isDone());
+ assertSame(testName, opend, future.get());
+
+ } catch (InterruptedException | ExecutionException e) {
+ throw new IllegalStateException(e);
+ }
+
+ if (expectedOperations > 0) {
+ assertEquals(testName, expectedSubRequestId, opend.getSubRequestId());
+ }
+ }
+
+ assertEquals(testName, expectedOperations, oper.getCount());
+ }
+
+ private class MyOper extends OperationPartial {
+ @Getter
+ private int count = 0;
+
+ @Setter
+ private boolean genException;
+
+ @Setter
+ private int maxFailures = 0;
+
+ @Setter
+ private CompletableFuture<OperationOutcome> guard;
+
+
+ public MyOper() {
+ super(OperationPartialTest.this.params, operator);
+ }
+
+ @Override
+ protected OperationOutcome doOperation(int attempt, OperationOutcome operation) {
+ ++count;
+ if (genException) {
+ throw new IllegalStateException(EXPECTED_EXCEPTION);
+ }
+
+ operation.setSubRequestId(String.valueOf(attempt));
+
+ if (count > maxFailures) {
+ operation.setResult(PolicyResult.SUCCESS);
+ } else {
+ operation.setResult(PolicyResult.FAILURE);
+ }
+
+ return operation;
+ }
+
+ @Override
+ protected CompletableFuture<OperationOutcome> startGuardAsync() {
+ return (guard != null ? guard : super.startGuardAsync());
+ }
+
+ @Override
+ protected long getRetryWaitMs() {
+ /*
+ * Sleep timers run in the background, but we want to control things via the
+ * "executor", thus we avoid sleep timers altogether by simply returning 0.
+ */
+ return 0L;
+ }
+ }
+
+ /**
+ * Executor that will run tasks until the queue is empty or a maximum number of tasks
+ * have been executed. Doesn't actually run anything until {@link #runAll()} is
+ * invoked.
+ */
+ private static class MyExec implements Executor {
+ private static final int MAX_TASKS = MAX_PARALLEL_REQUESTS * 100;
+
+ private Queue<Runnable> commands = new LinkedList<>();
+
+ public MyExec() {
+ // do nothing
+ }
+
+ public int getQueueLength() {
+ return commands.size();
+ }
+
+ @Override
+ public void execute(Runnable command) {
+ commands.add(command);
+ }
+
+ public boolean runAll() {
+ for (int count = 0; count < MAX_TASKS && !commands.isEmpty(); ++count) {
+ commands.remove().run();
+ }
+
+ return commands.isEmpty();
+ }
+ }
+}
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 21bc656..370426f 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
@@ -20,1271 +20,88 @@
package org.onap.policy.controlloop.actorserviceprovider.impl;
-import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
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.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;
-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;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-import lombok.Getter;
-import lombok.Setter;
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.Operation;
import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
-import org.onap.policy.controlloop.actorserviceprovider.pipeline.PipelineControllerFuture;
-import org.onap.policy.controlloop.policy.PolicyResult;
public class OperatorPartialTest {
- private static final int MAX_PARALLEL_REQUESTS = 10;
- private static final String EXPECTED_EXCEPTION = "expected exception";
private static final String ACTOR = "my-actor";
- private static final String OPERATOR = "my-operator";
- private static final String TARGET = "my-target";
- private static final int TIMEOUT = 1000;
- private static final UUID REQ_ID = UUID.randomUUID();
+ private static final String OPERATION = "my-name";
- private static final List<PolicyResult> FAILURE_RESULTS = Arrays.asList(PolicyResult.values()).stream()
- .filter(result -> result != PolicyResult.SUCCESS).collect(Collectors.toList());
-
- private VirtualControlLoopEvent event;
- private Map<String, Object> config;
- private ControlLoopEventContext context;
- private MyExec executor;
- private ControlLoopOperationParams params;
-
- private MyOper oper;
-
- private int numStart;
- private int numEnd;
-
- private Instant tstart;
-
- private OperationOutcome opstart;
- private OperationOutcome opend;
+ private OperatorPartial operator;
/**
- * Initializes the fields, including {@link #oper}.
+ * Initializes {@link #operator}.
*/
@Before
public void setUp() {
- event = new VirtualControlLoopEvent();
- event.setRequestId(REQ_ID);
-
- config = new TreeMap<>();
- context = new ControlLoopEventContext(event);
- executor = new MyExec();
-
- params = ControlLoopOperationParams.builder().completeCallback(this::completer).context(context)
- .executor(executor).actor(ACTOR).operation(OPERATOR).timeoutSec(TIMEOUT)
- .startCallback(this::starter).targetEntity(TARGET).build();
-
- oper = new MyOper();
- oper.configure(new TreeMap<>());
- oper.start();
-
- tstart = null;
-
- opstart = null;
- opend = null;
+ operator = new OperatorPartial(ACTOR, OPERATION) {
+ @Override
+ public Operation buildOperation(ControlLoopOperationParams params) {
+ return null;
+ }
+ };
}
@Test
public void testOperatorPartial_testGetActorName_testGetName() {
- assertEquals(ACTOR, oper.getActorName());
- assertEquals(OPERATOR, oper.getName());
- assertEquals(ACTOR + "." + OPERATOR, oper.getFullName());
- }
-
- @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);
+ assertEquals(ACTOR, operator.getActorName());
+ assertEquals(OPERATION, operator.getName());
+ assertEquals(ACTOR + "." + OPERATION, operator.getFullName());
}
@Test
public void testDoStart() {
- oper = spy(new MyOper());
+ operator.configure(null);
- oper.configure(config);
- oper.start();
+ operator = spy(operator);
+ operator.start();
- verify(oper).doStart();
-
- // others should not have been invoked
- verify(oper, never()).doStop();
- verify(oper, never()).doShutdown();
+ verify(operator).doStart();
}
@Test
public void testDoStop() {
- oper = spy(new MyOper());
+ operator.configure(null);
+ operator.start();
- oper.configure(config);
- oper.start();
- oper.stop();
+ operator = spy(operator);
+ operator.stop();
- verify(oper).doStop();
-
- // should not have been re-invoked
- verify(oper).doStart();
-
- // others should not have been invoked
- verify(oper, never()).doShutdown();
+ verify(operator).doStop();
}
@Test
public void testDoShutdown() {
- oper = spy(new MyOper());
+ operator.configure(null);
+ operator.start();
- oper.configure(config);
- oper.start();
- oper.shutdown();
+ operator = spy(operator);
+ operator.shutdown();
- verify(oper).doShutdown();
-
- // should not have been re-invoked
- verify(oper).doStart();
-
- // others should not have been invoked
- verify(oper, never()).doStop();
+ verify(operator).doShutdown();
}
@Test
- public void testStartOperation() {
- verifyRun("testStartOperation", 1, 1, PolicyResult.SUCCESS);
- }
+ public void testDoConfigureMapOfStringObject() {
+ operator = spy(operator);
- /**
- * Tests startOperation() when the operator is not running.
- */
- @Test
- public void testStartOperationNotRunning() {
- // use a new operator, one that hasn't been started yet
- oper = new MyOper();
- oper.configure(new TreeMap<>());
+ Map<String, Object> params = new TreeMap<>();
+ operator.configure(params);
- assertThatIllegalStateException().isThrownBy(() -> oper.startOperation(params));
- }
-
- /**
- * Tests startOperation() when the operation has a preprocessor.
- */
- @Test
- public void testStartOperationWithPreprocessor() {
- AtomicInteger count = new AtomicInteger();
-
- CompletableFuture<OperationOutcome> preproc = CompletableFuture.supplyAsync(() -> {
- count.incrementAndGet();
- return makeSuccess();
- }, executor);
-
- oper.setPreProcessor(preproc);
-
- verifyRun("testStartOperationWithPreprocessor_testStartPreprocessor", 1, 1, PolicyResult.SUCCESS);
-
- assertEquals(1, count.get());
- }
-
- /**
- * Tests startOperation() with multiple running requests.
- */
- @Test
- public void testStartOperationMultiple() {
- for (int count = 0; count < MAX_PARALLEL_REQUESTS; ++count) {
- oper.startOperation(params);
- }
-
- assertTrue(executor.runAll());
-
- assertNotNull(opstart);
- assertNotNull(opend);
- assertEquals(PolicyResult.SUCCESS, opend.getResult());
-
- assertEquals(MAX_PARALLEL_REQUESTS, numStart);
- assertEquals(MAX_PARALLEL_REQUESTS, oper.getCount());
- assertEquals(MAX_PARALLEL_REQUESTS, numEnd);
- }
-
- /**
- * Tests startPreprocessor() when the preprocessor returns a failure.
- */
- @Test
- public void testStartPreprocessorFailure() {
- oper.setPreProcessor(CompletableFuture.completedFuture(makeFailure()));
-
- verifyRun("testStartPreprocessorFailure", 1, 0, PolicyResult.FAILURE_GUARD);
- }
-
- /**
- * Tests startPreprocessor() when the preprocessor throws an exception.
- */
- @Test
- public void testStartPreprocessorException() {
- // arrange for the preprocessor to throw an exception
- oper.setPreProcessor(CompletableFuture.failedFuture(new IllegalStateException(EXPECTED_EXCEPTION)));
-
- verifyRun("testStartPreprocessorException", 1, 0, PolicyResult.FAILURE_GUARD);
- }
-
- /**
- * Tests startPreprocessor() when the pipeline is not running.
- */
- @Test
- public void testStartPreprocessorNotRunning() {
- // arrange for the preprocessor to return success, which will be ignored
- oper.setPreProcessor(CompletableFuture.completedFuture(makeSuccess()));
-
- oper.startOperation(params).cancel(false);
- assertTrue(executor.runAll());
-
- assertNull(opstart);
- assertNull(opend);
-
- assertEquals(0, numStart);
- assertEquals(0, oper.getCount());
- assertEquals(0, numEnd);
- }
-
- /**
- * Tests startPreprocessor() when the preprocessor <b>builder</b> throws an exception.
- */
- @Test
- public void testStartPreprocessorBuilderException() {
- oper = new MyOper() {
- @Override
- protected CompletableFuture<OperationOutcome> startPreprocessorAsync(ControlLoopOperationParams params) {
- 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());
+ verify(operator).doConfigure(params);
}
@Test
- public void testStartPreprocessorAsync() {
- assertNull(oper.startPreprocessorAsync(params));
- }
-
- @Test
- public void testStartOperationAsync() {
- oper.startOperation(params);
- assertTrue(executor.runAll());
-
- assertEquals(1, oper.getCount());
- }
-
- @Test
- public void testIsSuccess() {
- OperationOutcome outcome = new OperationOutcome();
-
- outcome.setResult(PolicyResult.SUCCESS);
- assertTrue(oper.isSuccess(outcome));
-
- for (PolicyResult failure : FAILURE_RESULTS) {
- outcome.setResult(failure);
- assertFalse("testIsSuccess-" + failure, oper.isSuccess(outcome));
- }
- }
-
- @Test
- public void testIsActorFailed() {
- assertFalse(oper.isActorFailed(null));
-
- OperationOutcome outcome = params.makeOutcome();
-
- // incorrect outcome
- outcome.setResult(PolicyResult.SUCCESS);
- assertFalse(oper.isActorFailed(outcome));
-
- outcome.setResult(PolicyResult.FAILURE_RETRIES);
- assertFalse(oper.isActorFailed(outcome));
-
- // correct outcome
- outcome.setResult(PolicyResult.FAILURE);
-
- // incorrect actor
- outcome.setActor(TARGET);
- assertFalse(oper.isActorFailed(outcome));
- outcome.setActor(null);
- assertFalse(oper.isActorFailed(outcome));
- outcome.setActor(ACTOR);
-
- // incorrect operation
- outcome.setOperation(TARGET);
- assertFalse(oper.isActorFailed(outcome));
- outcome.setOperation(null);
- assertFalse(oper.isActorFailed(outcome));
- outcome.setOperation(OPERATOR);
-
- // correct values
- assertTrue(oper.isActorFailed(outcome));
- }
-
- @Test
- public void testDoOperation() {
- /*
- * Use an operator that doesn't override doOperation().
- */
- OperatorPartial oper2 = new OperatorPartial(ACTOR, OPERATOR) {
- @Override
- protected Executor getBlockingExecutor() {
- return executor;
- }
- };
-
- oper2.configure(new TreeMap<>());
- oper2.start();
-
- oper2.startOperation(params);
- assertTrue(executor.runAll());
-
- assertNotNull(opend);
- assertEquals(PolicyResult.FAILURE_EXCEPTION, opend.getResult());
- }
-
- @Test
- public void testTimeout() throws Exception {
-
- // use a real executor
- params = params.toBuilder().executor(ForkJoinPool.commonPool()).build();
-
- // trigger timeout very quickly
- oper = new MyOper() {
- @Override
- protected long getTimeOutMillis(Integer timeoutSec) {
- return 1;
- }
-
- @Override
- 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, oper.startOperation(params).get().getResult());
- }
-
- /**
- * Verifies that the timer doesn't encompass the preprocessor and doesn't stop the
- * operation once the preprocessor completes.
- */
- @Test
- public void testTimeoutInPreprocessor() throws Exception {
-
- // use a real executor
- params = params.toBuilder().executor(ForkJoinPool.commonPool()).build();
-
- // trigger timeout very quickly
- oper = new MyOper() {
- @Override
- protected long getTimeOutMillis(Integer timeoutSec) {
- return 10;
- }
-
- @Override
- 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();
-
- OperationOutcome result = oper.startOperation(params).get();
- assertEquals(PolicyResult.SUCCESS, result.getResult());
-
- assertNotNull(opstart);
- assertNotNull(opend);
- assertEquals(PolicyResult.SUCCESS, opend.getResult());
-
- assertEquals(1, numStart);
- assertEquals(1, oper.getCount());
- assertEquals(1, numEnd);
- }
-
- /**
- * Tests retry functions, when the count is set to zero and retries are exhausted.
- */
- @Test
- public void testSetRetryFlag_testRetryOnFailure_ZeroRetries_testStartOperationAttempt() {
- params = params.toBuilder().retry(0).build();
- oper.setMaxFailures(10);
-
- verifyRun("testSetRetryFlag_testRetryOnFailure_ZeroRetries", 1, 1, PolicyResult.FAILURE);
- }
-
- /**
- * Tests retry functions, when the count is null and retries are exhausted.
- */
- @Test
- public void testSetRetryFlag_testRetryOnFailure_NullRetries() {
- params = params.toBuilder().retry(null).build();
- oper.setMaxFailures(10);
-
- verifyRun("testSetRetryFlag_testRetryOnFailure_NullRetries", 1, 1, PolicyResult.FAILURE);
- }
-
- /**
- * Tests retry functions, when retries are exhausted.
- */
- @Test
- public void testSetRetryFlag_testRetryOnFailure_RetriesExhausted() {
- final int maxRetries = 3;
- params = params.toBuilder().retry(maxRetries).build();
- oper.setMaxFailures(10);
-
- verifyRun("testSetRetryFlag_testRetryOnFailure_RetriesExhausted", maxRetries + 1, maxRetries + 1,
- PolicyResult.FAILURE_RETRIES);
- }
-
- /**
- * Tests retry functions, when a success follows some retries.
- */
- @Test
- public void testSetRetryFlag_testRetryOnFailure_SuccessAfterRetries() {
- params = params.toBuilder().retry(10).build();
-
- final int maxFailures = 3;
- oper.setMaxFailures(maxFailures);
-
- verifyRun("testSetRetryFlag_testRetryOnFailure_SuccessAfterRetries", maxFailures + 1, maxFailures + 1,
- PolicyResult.SUCCESS);
- }
-
- /**
- * Tests retry functions, when the outcome is {@code null}.
- */
- @Test
- public void testSetRetryFlag_testRetryOnFailure_NullOutcome() {
-
- // arrange to return null from doOperation()
- oper = new MyOper() {
- @Override
- protected OperationOutcome doOperation(ControlLoopOperationParams params, int attempt,
- OperationOutcome operation) {
-
- // update counters
- super.doOperation(params, attempt, operation);
- return null;
- }
- };
-
- oper.configure(new TreeMap<>());
- oper.start();
-
- verifyRun("testSetRetryFlag_testRetryOnFailure_NullOutcome", 1, 1, PolicyResult.FAILURE, null, noop());
- }
-
- @Test
- public void testIsSameOperation() {
- assertFalse(oper.isSameOperation(null));
-
- OperationOutcome outcome = params.makeOutcome();
-
- // wrong actor - should be false
- outcome.setActor(null);
- assertFalse(oper.isSameOperation(outcome));
- outcome.setActor(TARGET);
- assertFalse(oper.isSameOperation(outcome));
- outcome.setActor(ACTOR);
-
- // wrong operation - should be null
- outcome.setOperation(null);
- assertFalse(oper.isSameOperation(outcome));
- outcome.setOperation(TARGET);
- assertFalse(oper.isSameOperation(outcome));
- outcome.setOperation(OPERATOR);
-
- assertTrue(oper.isSameOperation(outcome));
- }
-
- /**
- * Tests handleFailure() when the outcome is a success.
- */
- @Test
- public void testHandlePreprocessorFailureTrue() {
- oper.setPreProcessor(CompletableFuture.completedFuture(makeSuccess()));
- verifyRun("testHandlePreprocessorFailureTrue", 1, 1, PolicyResult.SUCCESS);
- }
-
- /**
- * Tests handleFailure() when the outcome is <i>not</i> a success.
- */
- @Test
- public void testHandlePreprocessorFailureFalse() throws Exception {
- oper.setPreProcessor(CompletableFuture.completedFuture(makeFailure()));
- verifyRun("testHandlePreprocessorFailureFalse", 1, 0, PolicyResult.FAILURE_GUARD);
- }
-
- /**
- * Tests handleFailure() when the outcome is {@code null}.
- */
- @Test
- public void testHandlePreprocessorFailureNull() throws Exception {
- // arrange to return null from the preprocessor
- oper.setPreProcessor(CompletableFuture.completedFuture(null));
-
- verifyRun("testHandlePreprocessorFailureNull", 1, 0, PolicyResult.FAILURE_GUARD);
- }
-
- @Test
- public void testFromException() {
- // arrange to generate an exception when operation runs
- oper.setGenException(true);
-
- verifyRun("testFromException", 1, 1, PolicyResult.FAILURE_EXCEPTION);
- }
-
- /**
- * Tests fromException() when there is no exception.
- */
- @Test
- public void testFromExceptionNoExcept() {
- verifyRun("testFromExceptionNoExcept", 1, 1, PolicyResult.SUCCESS);
- }
-
- /**
- * 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 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());
- }
-
- /**
- * Tests callbackStarted() when the pipeline has already been stopped.
- */
- @Test
- public void testCallbackStartedNotRunning() {
- AtomicReference<Future<OperationOutcome>> future = new AtomicReference<>();
-
- /*
- * arrange to stop the controller when the start-callback is invoked, but capture
- * the outcome
- */
- params = params.toBuilder().startCallback(oper -> {
- starter(oper);
- future.get().cancel(false);
- }).build();
-
- future.set(oper.startOperation(params));
- assertTrue(executor.runAll());
-
- // should have only run once
- assertEquals(1, numStart);
- }
-
- /**
- * Tests callbackCompleted() when the pipeline has already been stopped.
- */
- @Test
- public void testCallbackCompletedNotRunning() {
- AtomicReference<Future<OperationOutcome>> future = new AtomicReference<>();
-
- // arrange to stop the controller when the start-callback is invoked
- params = params.toBuilder().startCallback(oper -> {
- future.get().cancel(false);
- }).build();
-
- future.set(oper.startOperation(params));
- assertTrue(executor.runAll());
-
- // should not have been set
- assertNull(opend);
- assertEquals(0, numEnd);
- }
-
- @Test
- public void testSetOutcomeControlLoopOperationOutcomeThrowable() {
- final CompletionException timex = new CompletionException(new TimeoutException(EXPECTED_EXCEPTION));
-
- OperationOutcome outcome;
-
- outcome = new OperationOutcome();
- oper.setOutcome(params, outcome, timex);
- assertEquals(ControlLoopOperation.FAILED_MSG, outcome.getMessage());
- assertEquals(PolicyResult.FAILURE_TIMEOUT, outcome.getResult());
-
- outcome = new OperationOutcome();
- oper.setOutcome(params, outcome, new IllegalStateException(EXPECTED_EXCEPTION));
- assertEquals(ControlLoopOperation.FAILED_MSG, outcome.getMessage());
- assertEquals(PolicyResult.FAILURE_EXCEPTION, outcome.getResult());
- }
-
- @Test
- public void testSetOutcomeControlLoopOperationOutcomePolicyResult() {
- OperationOutcome outcome;
-
- outcome = new OperationOutcome();
- oper.setOutcome(params, outcome, PolicyResult.SUCCESS);
- assertEquals(ControlLoopOperation.SUCCESS_MSG, outcome.getMessage());
- assertEquals(PolicyResult.SUCCESS, outcome.getResult());
-
- for (PolicyResult result : FAILURE_RESULTS) {
- outcome = new OperationOutcome();
- oper.setOutcome(params, outcome, result);
- assertEquals(result.toString(), ControlLoopOperation.FAILED_MSG, outcome.getMessage());
- assertEquals(result.toString(), result, outcome.getResult());
- }
- }
-
- @Test
- public void testIsTimeout() {
- final TimeoutException timex = new TimeoutException(EXPECTED_EXCEPTION);
-
- 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)));
- assertFalse(oper.isTimeout(new CompletionException(new CompletionException(timex))));
-
- assertTrue(oper.isTimeout(timex));
- assertTrue(oper.isTimeout(new CompletionException(timex)));
- }
-
- @Test
- public void testGetTimeOutMillis() {
- assertEquals(TIMEOUT * 1000, oper.getTimeOutMillis(params.getTimeoutSec()));
-
- params = params.toBuilder().timeoutSec(null).build();
- assertEquals(0, oper.getTimeOutMillis(params.getTimeoutSec()));
- }
-
- private void starter(OperationOutcome oper) {
- ++numStart;
- tstart = oper.getStart();
- opstart = oper;
- }
-
- private void completer(OperationOutcome oper) {
- ++numEnd;
- opend = oper;
- }
-
- /**
- * Gets a function that does nothing.
- *
- * @param <T> type of input parameter expected by the function
- * @return a function that does nothing
- */
- private <T> Consumer<T> noop() {
- return unused -> {
- };
- }
-
- private OperationOutcome makeSuccess() {
- OperationOutcome outcome = params.makeOutcome();
- outcome.setResult(PolicyResult.SUCCESS);
-
- return outcome;
- }
-
- private OperationOutcome makeFailure() {
- OperationOutcome outcome = params.makeOutcome();
- outcome.setResult(PolicyResult.FAILURE);
-
- return outcome;
- }
-
- /**
- * 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) {
-
- String expectedSubRequestId =
- (expectedResult == PolicyResult.FAILURE_EXCEPTION ? null : String.valueOf(expectedOperations));
-
- verifyRun(testName, expectedCallbacks, expectedOperations, expectedResult, expectedSubRequestId, noop());
- }
-
- /**
- * Verifies a run.
- *
- * @param testName test name
- * @param expectedCallbacks number of callbacks expected
- * @param expectedOperations number of operation invocations expected
- * @param expectedResult expected outcome
- * @param expectedSubRequestId expected sub request ID
- * @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,
- String expectedSubRequestId, Consumer<CompletableFuture<OperationOutcome>> manipulator) {
-
- CompletableFuture<OperationOutcome> future = oper.startOperation(params);
-
- manipulator.accept(future);
-
- assertTrue(testName, executor.runAll());
-
- assertEquals(testName, expectedCallbacks, numStart);
- assertEquals(testName, expectedCallbacks, numEnd);
-
- if (expectedCallbacks > 0) {
- assertNotNull(testName, opstart);
- assertNotNull(testName, opend);
- assertEquals(testName, expectedResult, opend.getResult());
-
- assertSame(testName, tstart, opstart.getStart());
- assertSame(testName, tstart, opend.getStart());
-
- try {
- assertTrue(future.isDone());
- assertSame(testName, opend, future.get());
-
- } catch (InterruptedException | ExecutionException e) {
- throw new IllegalStateException(e);
- }
-
- if (expectedOperations > 0) {
- assertEquals(testName, expectedSubRequestId, opend.getSubRequestId());
- }
- }
-
- assertEquals(testName, expectedOperations, oper.getCount());
- }
-
- private class MyOper extends OperatorPartial {
- @Getter
- private int count = 0;
-
- @Setter
- private boolean genException;
-
- @Setter
- private int maxFailures = 0;
-
- @Setter
- private CompletableFuture<OperationOutcome> preProcessor;
-
- public MyOper() {
- super(ACTOR, OPERATOR);
- }
-
- @Override
- protected OperationOutcome doOperation(ControlLoopOperationParams params, int attempt,
- OperationOutcome operation) {
- ++count;
- if (genException) {
- throw new IllegalStateException(EXPECTED_EXCEPTION);
- }
-
- operation.setSubRequestId(String.valueOf(attempt));
-
- if (count > maxFailures) {
- operation.setResult(PolicyResult.SUCCESS);
- } else {
- operation.setResult(PolicyResult.FAILURE);
- }
-
- return operation;
- }
-
- @Override
- protected CompletableFuture<OperationOutcome> startPreprocessorAsync(ControlLoopOperationParams params) {
- return (preProcessor != null ? preProcessor : super.startPreprocessorAsync(params));
- }
-
- @Override
- protected Executor getBlockingExecutor() {
- return executor;
- }
- }
-
- /**
- * Executor that will run tasks until the queue is empty or a maximum number of tasks
- * have been executed.
- */
- private static class MyExec implements Executor {
- private static final int MAX_TASKS = MAX_PARALLEL_REQUESTS * 100;
-
- private Queue<Runnable> commands = new LinkedList<>();
-
- public MyExec() {
- // do nothing
- }
-
- public int getQueueLength() {
- return commands.size();
- }
-
- @Override
- public void execute(Runnable command) {
- commands.add(command);
- }
-
- public boolean runAll() {
- for (int count = 0; count < MAX_TASKS && !commands.isEmpty(); ++count) {
- commands.remove().run();
- }
-
- return commands.isEmpty();
- }
+ public void testGetBlockingExecutor() {
+ assertNotNull(operator.getBlockingExecutor());
}
}
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 9dd19d5..a5215a4 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
@@ -51,6 +51,7 @@
import org.onap.policy.common.parameters.BeanValidationResult;
import org.onap.policy.controlloop.VirtualControlLoopEvent;
import org.onap.policy.controlloop.actorserviceprovider.ActorService;
+import org.onap.policy.controlloop.actorserviceprovider.Operation;
import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
import org.onap.policy.controlloop.actorserviceprovider.Operator;
import org.onap.policy.controlloop.actorserviceprovider.controlloop.ControlLoopEventContext;
@@ -87,12 +88,15 @@
private Executor executor;
@Mock
- private CompletableFuture<OperationOutcome> operation;
+ private CompletableFuture<OperationOutcome> operFuture;
@Mock
private Operator operator;
@Mock
+ private Operation operation;
+
+ @Mock
private Consumer<OperationOutcome> starter;
private Map<String, String> payload;
@@ -110,7 +114,8 @@
when(actorService.getActor(ACTOR)).thenReturn(actor);
when(actor.getOperator(OPERATION)).thenReturn(operator);
- when(operator.startOperation(any())).thenReturn(operation);
+ when(operator.buildOperation(any())).thenReturn(operation);
+ when(operation.start()).thenReturn(operFuture);
when(event.getRequestId()).thenReturn(REQ_ID);
@@ -128,7 +133,7 @@
@Test
public void testStart() {
- assertSame(operation, params.start());
+ assertSame(operFuture, params.start());
assertThatIllegalArgumentException().isThrownBy(() -> params.toBuilder().context(null).build().start());
}
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 6c1f538..daa0aff 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
@@ -41,7 +41,7 @@
private static final String CONTAINER = "my-container";
private static final String CLIENT = "my-client";
- private static final long TIMEOUT = 10;
+ private static final int TIMEOUT = 10;
private static final String PATH1 = "path #1";
private static final String PATH2 = "path #2";
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 6cf7328..ae4a79f 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
@@ -36,7 +36,7 @@
private static final String CONTAINER = "my-container";
private static final String CLIENT = "my-client";
private static final String PATH = "my-path";
- private static final long TIMEOUT = 10;
+ private static final int TIMEOUT = 10;
private HttpParams params;
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 a6b11ef..4a00c06 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
@@ -424,7 +424,7 @@
}
/**
- * Tests add(Function) when the controller is canceled after the future is added.
+ * Tests wrap(Function) when the controller is canceled after the future is added.
*/
@Test
public void testWrapFunctionCancel() throws Exception {
@@ -442,7 +442,7 @@
}
/**
- * Tests add(Function) when the controller is not running.
+ * Tests wrap(Function) when the controller is not running.
*/
@Test
public void testWrapFunctionNotRunning() {
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 c7fe46e..8604688 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
@@ -39,4 +39,9 @@
<logger name="org.onap.policy.controlloop.actorserviceprovider.Util" level="info" additivity="false">
<appender-ref ref="STDOUT" />
</logger>
+
+ <!-- this is required for HttpOperationTest -->
+ <logger name="org.onap.policy.controlloop.actorserviceprovider.impl.HttpOperation" level="info" additivity="false">
+ <appender-ref ref="STDOUT" />
+ </logger>
</configuration>
diff --git a/models-interactions/model-actors/pom.xml b/models-interactions/model-actors/pom.xml
index 8765eb4..029ac7f 100644
--- a/models-interactions/model-actors/pom.xml
+++ b/models-interactions/model-actors/pom.xml
@@ -36,6 +36,7 @@
<modules>
<module>actorServiceProvider</module>
+ <module>actor.test</module>
<module>actor.appc</module>
<module>actor.vfc</module>
<module>actor.sdnc</module>
diff --git a/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/AaiConstants.java b/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/AaiConstants.java
new file mode 100644
index 0000000..084e4a5
--- /dev/null
+++ b/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/AaiConstants.java
@@ -0,0 +1,34 @@
+/*-
+ * ============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.aai;
+
+/**
+ * Constants used with A&AI classes.
+ */
+public class AaiConstants {
+
+ public static final String ACTOR_NAME = "AAI";
+ public static final String CONTEXT_PREFIX = ACTOR_NAME + ".";
+
+ private AaiConstants() {
+ // do nothing
+ }
+}
diff --git a/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/AaiCqResponse.java b/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/AaiCqResponse.java
index 2010a5d..6fb42db 100644
--- a/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/AaiCqResponse.java
+++ b/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/AaiCqResponse.java
@@ -2,7 +2,7 @@
* ============LICENSE_START=======================================================
*
* ================================================================================
- * Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2019-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.
@@ -50,6 +50,7 @@
public class AaiCqResponse implements Serializable {
private static final long serialVersionUID = 1L;
+ public static final String CONTEXT_KEY = AaiConstants.CONTEXT_PREFIX + "AaiCqResponse";
private static final String GENERIC_VNF = "generic-vnf";
private static final String VF_MODULE = "vf-module";
private static final Logger LOGGER = LoggerFactory.getLogger(AaiCqResponse.class);
diff --git a/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/AaiManager.java b/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/AaiManager.java
index 923b8d3..41f4d3b 100644
--- a/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/AaiManager.java
+++ b/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/AaiManager.java
@@ -2,7 +2,7 @@
* ============LICENSE_START=======================================================
* aai
* ================================================================================
- * Copyright (C) 2017-2019 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2017-2020 AT&T Intellectual Property. All rights reserved.
* Modifications Copyright (C) 2019 Nordix Foundation.
* Modifications Copyright (C) 2019 Samsung Electronics Co., Ltd.
* ================================================================================
@@ -47,6 +47,8 @@
*/
public final class AaiManager {
+ // TODO remove this class
+
/** The Constant logger. */
private static final Logger logger = LoggerFactory.getLogger(AaiManager.class);
diff --git a/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/util/AaiException.java b/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/util/AaiException.java
index 1fe23cf..83923f1 100644
--- a/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/util/AaiException.java
+++ b/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/util/AaiException.java
@@ -23,6 +23,8 @@
public class AaiException extends Exception {
+ // TODO remove this class
+
private static final long serialVersionUID = 9220983727706207465L;
public AaiException() {
diff --git a/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/util/Serialization.java b/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/util/Serialization.java
index e42325f..326294a 100644
--- a/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/util/Serialization.java
+++ b/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/util/Serialization.java
@@ -26,6 +26,8 @@
public final class Serialization {
+ // TODO remove this class
+
public static final Gson gsonPretty = new GsonBuilder().disableHtmlEscaping().setPrettyPrinting().create();
private Serialization() {