summaryrefslogtreecommitdiffstats
path: root/models-interactions/model-actors/actorServiceProvider
diff options
context:
space:
mode:
Diffstat (limited to 'models-interactions/model-actors/actorServiceProvider')
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/AsyncResponseHandler.java119
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/Operation.java52
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/Operator.java10
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/Util.java79
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/controlloop/ControlLoopEventContext.java45
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/ActorImpl.java2
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperation.java286
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperator.java44
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperationPartial.java844
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperatorPartial.java758
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/parameters/ControlLoopOperationParams.java3
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpActorParams.java2
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpParams.java2
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/spi/Actor.java1
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/AsyncResponseHandlerTest.java172
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/UtilTest.java82
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/controlloop/ControlLoopEventContextTest.java43
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/ActorImplTest.java9
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpActorTest.java2
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperationTest.java781
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperatorTest.java118
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperationPartialTest.java1302
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperatorPartialTest.java1251
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/ControlLoopOperationParamsTest.java11
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpActorParamsTest.java2
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpParamsTest.java2
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/PipelineControllerFutureTest.java4
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/resources/logback-test.xml5
28 files changed, 3549 insertions, 2482 deletions
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 d78403809..000000000
--- 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 000000000..39977fd41
--- /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 c09460e34..24faafd40 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 @@ public interface Operator extends Startable, Configurable<Map<String, Object>> {
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 c3ddd17f3..b885b5c25 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 @@ package org.onap.policy.controlloop.actorserviceprovider;
import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.Map;
-import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
-import org.onap.policy.common.endpoints.utils.NetLoggerUtil;
-import org.onap.policy.common.endpoints.utils.NetLoggerUtil.EventType;
import org.onap.policy.common.utils.coder.Coder;
import org.onap.policy.common.utils.coder.CoderException;
import org.onap.policy.common.utils.coder.StandardCoder;
@@ -56,82 +53,6 @@ public class Util {
}
/**
- * Logs a REST request. If the request is not of type, String, then it attempts to
- * pretty-print it into JSON before logging.
- *
- * @param url request URL
- * @param request request to be logged
- */
- public static <T> void logRestRequest(String url, T request) {
- logRestRequest(new StandardCoder(), url, request);
- }
-
- /**
- * Logs a REST request. If the request is not of type, String, then it attempts to
- * pretty-print it into JSON before logging.
- *
- * @param coder coder to be used to pretty-print the request
- * @param url request URL
- * @param request request to be logged
- */
- protected static <T> void logRestRequest(Coder coder, String url, T request) {
- String json;
- try {
- if (request instanceof String) {
- json = request.toString();
- } else {
- json = coder.encode(request, true);
- }
-
- } catch (CoderException e) {
- logger.warn("cannot pretty-print request", e);
- json = request.toString();
- }
-
- NetLoggerUtil.log(EventType.OUT, CommInfrastructure.REST, url, json);
- logger.info("[OUT|{}|{}|]{}{}", CommInfrastructure.REST, url, NetLoggerUtil.SYSTEM_LS, json);
- }
-
- /**
- * Logs a REST response. If the response is not of type, String, then it attempts to
- * pretty-print it into JSON before logging.
- *
- * @param url request URL
- * @param response response to be logged
- */
- public static <T> void logRestResponse(String url, T response) {
- logRestResponse(new StandardCoder(), url, response);
- }
-
- /**
- * Logs a REST response. If the request is not of type, String, then it attempts to
- * pretty-print it into JSON before logging.
- *
- * @param coder coder to be used to pretty-print the response
- * @param url request URL
- * @param response response to be logged
- */
- protected static <T> void logRestResponse(Coder coder, String url, T response) {
- String json;
- try {
- if (response == null) {
- json = null;
- } else if (response instanceof String) {
- json = response.toString();
- } else {
- json = coder.encode(response, true);
- }
-
- } catch (CoderException e) {
- logger.warn("cannot pretty-print response", e);
- json = response.toString();
- }
-
- NetLoggerUtil.log(EventType.IN, CommInfrastructure.REST, url, json);
- logger.info("[IN|{}|{}|]{}{}", CommInfrastructure.REST, url, NetLoggerUtil.SYSTEM_LS, json);
- }
-
- /**
* Runs a function and logs a message if it throws an exception. Does <i>not</i>
* re-throw the exception.
*
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 cd4d2570f..1c37a8e0d 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 @@ package org.onap.policy.controlloop.actorserviceprovider.controlloop;
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 @@ public class ControlLoopEventContext implements Serializable {
*/
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 class ControlLoopEventContext implements Serializable {
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 d7f322e8a..0c88ebee2 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 class ActorImpl extends StartConfigPartial<Map<String, Object>> implement
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 000000000..c4bf5f484
--- /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 566492907..add74aa42 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 @@ public class HttpOperator extends OperatorPartial {
}
/**
+ * 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 @@ public class HttpOperator extends OperatorPartial {
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 000000000..d00b88bb5
--- /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 df5258d71..3e15c1be4 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 @@ public abstract class OperatorPartial extends StartConfigPartial<Map<String, Obj
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 57fce40d7..925916097 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 @@ public class ControlLoopOperationParams {
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 da4fb4f0c..275c8bc4e 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 @@ public class HttpActorParams {
* 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 695ffe4dd..93711c032 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 @@ public class HttpParams {
*/
@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 620950a3c..53bee5f00 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 31c6d2077..000000000
--- 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 4a3f321cf..0a2a5a90e 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.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
-import org.onap.policy.common.utils.coder.CoderException;
-import org.onap.policy.common.utils.coder.StandardCoder;
import org.onap.policy.common.utils.test.log.logback.ExtractAppender;
import org.slf4j.LoggerFactory;
public class UtilTest {
- private static final String MY_REQUEST = "my-request";
- private static final String URL = "my-url";
- private static final String OUT_URL = "OUT|REST|my-url";
- private static final String IN_URL = "IN|REST|my-url";
protected static final String EXPECTED_EXCEPTION = "expected exception";
/**
@@ -89,82 +83,6 @@ public class UtilTest {
}
@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 0d917ad3e..b462043d5 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.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.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 @@ public class ControlLoopEventContextTest {
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 a209fb0d8..92cbbe774 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.Before;
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 @@ public class ActorImplTest {
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 2da789989..8ce3b3230 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 @@ public class HttpActorTest {
private static final String ACTOR = "my-actor";
private static final String UNKNOWN = "unknown";
private static final String CLIENT = "my-client";
- private static final long TIMEOUT = 10L;
+ private 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 000000000..19f781d61
--- /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 c006cf333..081bb346b 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 @@ package org.onap.policy.controlloop.actorserviceprovider.impl;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.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,62 +49,116 @@ public class HttpOperatorTest {
private static final String ACTOR = "my-actor";
private static final String OPERATION = "my-name";
- private static final String CLIENT = "my-client";
- private static final String PATH = "my-path";
- private static final long TIMEOUT = 100;
+ 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);
+
+ oper = new MyOperator();
+
+ HttpParams params = HttpParams.builder().clientName(HTTP_CLIENT).path(PATH).timeoutSec(TIMEOUT).build();
+ Map<String, Object> paramMap = Util.translateToMap(OPERATION, params);
+ oper.configure(paramMap);
}
@Test
- public void testDoConfigureMapOfStringObject_testGetClient_testGetPath_testGetTimeoutSec() {
+ public void testHttpOperator() {
+ assertEquals(ACTOR, oper.getActorName());
+ assertEquals(OPERATION, oper.getName());
+ assertEquals(ACTOR + "." + OPERATION, oper.getFullName());
+ }
+
+ @Test
+ public void testGetClient() {
+ 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());
- assertEquals(0L, oper.getTimeoutSec());
-
- oper = new HttpOperator(ACTOR, OPERATION) {
- @Override
- protected HttpClientFactory getClientFactory() {
- HttpClientFactory factory = mock(HttpClientFactory.class);
- when(factory.get(CLIENT)).thenReturn(client);
- return factory;
- }
- };
-
- HttpParams params = HttpParams.builder().clientName(CLIENT).path(PATH).timeoutSec(TIMEOUT).build();
+
+ // 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());
- assertEquals(TIMEOUT, oper.getTimeoutSec());
+
+ // should use given value
+ assertEquals(TIMEOUT * 1000, oper.getTimeoutMs());
// test invalid parameters
paramMap.remove("path");
assertThatThrownBy(() -> oper.configure(paramMap)).isInstanceOf(ParameterValidationRuntimeException.class);
}
- @Test
- public void testHttpOperator() {
- assertEquals(ACTOR, oper.getActorName());
- assertEquals(OPERATION, oper.getName());
- assertEquals(ACTOR + "." + OPERATION, oper.getFullName());
+ private class MyOperator extends HttpOperator {
+ public MyOperator() {
+ super(ACTOR, OPERATION);
+ }
+
+ @Override
+ public Operation buildOperation(ControlLoopOperationParams params) {
+ return null;
+ }
+
+ @Override
+ protected HttpClientFactory getClientFactory() {
+ return factory;
+ }
}
- @Test
- public void testGetClient() {
- assertNotNull(oper.getClientFactory());
+ 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 000000000..0d5cb2444
--- /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 21bc656f2..370426fd4 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;
- }
-
- @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);
- }
-
- @Test
- public void testDoStart() {
- oper = spy(new MyOper());
-
- oper.configure(config);
- oper.start();
-
- verify(oper).doStart();
-
- // others should not have been invoked
- verify(oper, never()).doStop();
- verify(oper, never()).doShutdown();
- }
-
- @Test
- public void testDoStop() {
- oper = spy(new MyOper());
-
- oper.configure(config);
- oper.start();
- oper.stop();
-
- verify(oper).doStop();
-
- // should not have been re-invoked
- verify(oper).doStart();
-
- // others should not have been invoked
- verify(oper, never()).doShutdown();
- }
-
- @Test
- public void testDoShutdown() {
- oper = spy(new MyOper());
-
- oper.configure(config);
- oper.start();
- oper.shutdown();
-
- verify(oper).doShutdown();
-
- // should not have been re-invoked
- verify(oper).doStart();
-
- // others should not have been invoked
- verify(oper, never()).doStop();
- }
-
- @Test
- public void testStartOperation() {
- verifyRun("testStartOperation", 1, 1, PolicyResult.SUCCESS);
- }
-
- /**
- * 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<>());
-
- 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());
- }
-
- @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() {
+ operator = new OperatorPartial(ACTOR, OPERATION) {
@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);
+ public Operation buildOperation(ControlLoopOperationParams params) {
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);
+ public void testOperatorPartial_testGetActorName_testGetName() {
+ assertEquals(ACTOR, operator.getActorName());
+ assertEquals(OPERATION, operator.getName());
+ assertEquals(ACTOR + "." + OPERATION, operator.getFullName());
}
- /**
- * 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();
+ public void testDoStart() {
+ operator.configure(null);
- future.set(oper.startOperation(params));
- assertTrue(executor.runAll());
+ operator = spy(operator);
+ operator.start();
- // should not have been set
- assertNull(opend);
- assertEquals(0, numEnd);
+ verify(operator).doStart();
}
@Test
- public void testSetOutcomeControlLoopOperationOutcomeThrowable() {
- final CompletionException timex = new CompletionException(new TimeoutException(EXPECTED_EXCEPTION));
-
- OperationOutcome outcome;
+ public void testDoStop() {
+ operator.configure(null);
+ operator.start();
- outcome = new OperationOutcome();
- oper.setOutcome(params, outcome, timex);
- assertEquals(ControlLoopOperation.FAILED_MSG, outcome.getMessage());
- assertEquals(PolicyResult.FAILURE_TIMEOUT, outcome.getResult());
+ operator = spy(operator);
+ operator.stop();
- outcome = new OperationOutcome();
- oper.setOutcome(params, outcome, new IllegalStateException(EXPECTED_EXCEPTION));
- assertEquals(ControlLoopOperation.FAILED_MSG, outcome.getMessage());
- assertEquals(PolicyResult.FAILURE_EXCEPTION, outcome.getResult());
+ verify(operator).doStop();
}
@Test
- public void testSetOutcomeControlLoopOperationOutcomePolicyResult() {
- OperationOutcome outcome;
+ public void testDoShutdown() {
+ operator.configure(null);
+ operator.start();
- outcome = new OperationOutcome();
- oper.setOutcome(params, outcome, PolicyResult.SUCCESS);
- assertEquals(ControlLoopOperation.SUCCESS_MSG, outcome.getMessage());
- assertEquals(PolicyResult.SUCCESS, outcome.getResult());
+ operator = spy(operator);
+ operator.shutdown();
- 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());
- }
+ verify(operator).doShutdown();
}
@Test
- public void testIsTimeout() {
- final TimeoutException timex = new TimeoutException(EXPECTED_EXCEPTION);
+ public void testDoConfigureMapOfStringObject() {
+ operator = spy(operator);
- 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))));
+ Map<String, Object> params = new TreeMap<>();
+ operator.configure(params);
- assertTrue(oper.isTimeout(timex));
- assertTrue(oper.isTimeout(new CompletionException(timex)));
+ verify(operator).doConfigure(params);
}
@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 9dd19d548..a5215a48f 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.mockito.MockitoAnnotations;
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 @@ public class ControlLoopOperationParamsTest {
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 @@ public class ControlLoopOperationParamsTest {
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 @@ public class ControlLoopOperationParamsTest {
@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 6c1f538ec..daa0affec 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 @@ public class HttpActorParamsTest {
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 6cf7328ca..ae4a79fe2 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 @@ public class HttpParamsTest {
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 a6b11ef65..4a00c065e 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 @@ public class PipelineControllerFutureTest {
}
/**
- * 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 @@ public class PipelineControllerFutureTest {
}
/**
- * 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 c7fe46e47..860468821 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>