Merge "More changes to actor code"

6 files changed, 63 insertions, 29 deletions

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 b885b5c25..ba4785922 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
@@ -34,6 +34,7 @@ import org.slf4j.LoggerFactory;
  */
 public class Util {
     private static final Logger logger = LoggerFactory.getLogger(Util.class);
+    private static final Coder coder = new StandardCoder();
 
     private Util() {
         // do nothing
@@ -84,11 +85,8 @@ public class Util {
      * @return the translated object
      */
     public static <T> T translate(String identifier, Object source, Class<T> clazz) {
-        Coder coder = new StandardCoder();
-
         try {
-            String json = coder.encode(source);
-            return coder.decode(json, clazz);
+            return coder.convert(source, clazz);
 
         } catch (CoderException | RuntimeException e) {
             throw new IllegalArgumentException("cannot translate parameters for " + identifier, e);
@@ -105,10 +103,6 @@ public class Util {
      */
     @SuppressWarnings("unchecked")
     public static Map<String, Object> translateToMap(String identifier, Object source) {
-        if (source == null) {
-            return null;
-        }
-
         return translate(identifier, source, LinkedHashMap.class);
     }
 }
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 1c37a8e0d..3e02da611 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
@@ -130,15 +130,27 @@ public class ControlLoopEventContext implements Serializable {
             return null;
         }
 
-        CompletableFuture<OperationOutcome> future = retrievers.get(name);
-        if (future != null) {
-            return future;
-        }
+        /*
+         * Return any existing future, if it wasn't canceled. Otherwise, start a new
+         * request.
+         */
 
-        future = params.start();
+        // @formatter:off
+        CompletableFuture<OperationOutcome> oldFuture =
+            retrievers.compute(name, (key, future) -> (future == null || future.isCancelled() ? null : future));
+        // @formatter:on
 
-        CompletableFuture<OperationOutcome> oldFuture = retrievers.putIfAbsent(name, future);
         if (oldFuture != null) {
+            return oldFuture;
+        }
+
+        /*
+         * Note: must NOT invoke params.start() within retrievers.compute(), as start()
+         * may invoke obtain() which would cause a recursive update to the retrievers map.
+         */
+        CompletableFuture<OperationOutcome> future = params.start();
+
+        if ((oldFuture = retrievers.putIfAbsent(name, future)) != null) {
             future.cancel(false);
             return oldFuture;
         }
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperation.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperation.java
index f82015d6b..d1e21f8fd 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperation.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperation.java
@@ -214,14 +214,14 @@ public abstract class BidirectionalTopicOperation<Q, S> extends OperationPartial
             case SUCCESS:
                 logger.info("{}.{} request succeeded for {}", params.getActor(), params.getOperation(),
                                 params.getRequestId());
-                setOutcome(outcome, PolicyResult.SUCCESS);
+                setOutcome(outcome, PolicyResult.SUCCESS, response);
                 postProcessResponse(outcome, rawResponse, response);
                 return outcome;
 
             case FAILURE:
                 logger.info("{}.{} request failed for {}", params.getActor(), params.getOperation(),
                                 params.getRequestId());
-                return setOutcome(outcome, PolicyResult.FAILURE);
+                return setOutcome(outcome, PolicyResult.FAILURE, response);
 
             case STILL_WAITING:
             default:
@@ -232,6 +232,18 @@ public abstract class BidirectionalTopicOperation<Q, S> extends OperationPartial
     }
 
     /**
+     * Sets an operation's outcome and default message based on the result.
+     *
+     * @param outcome operation to be updated
+     * @param result result of the operation
+     * @param response response used to populate the outcome
+     * @return the updated operation
+     */
+    public OperationOutcome setOutcome(OperationOutcome outcome, PolicyResult result, S response) {
+        return setOutcome(outcome, result);
+    }
+
+    /**
      * Processes a successful response.
      *
      * @param outcome outcome to be populated
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
index f1829d79a..c3c0f6dc2 100644
--- 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
@@ -148,21 +148,23 @@ public abstract class HttpOperation<T> extends OperationPartial {
         controller.add(requester.apply(callback));
 
         // once "future" completes, process the response, and then complete the controller
-        future.thenApplyAsync(response -> processResponse(outcome, url, response), executor)
+        future.thenComposeAsync(response -> processResponse(outcome, url, response), executor)
                         .whenCompleteAsync(controller.delayedComplete(), executor);
 
         return controller;
     }
 
     /**
-     * Processes a response. This method simply sets the outcome to SUCCESS.
+     * Processes a response. This method decodes the response, sets the outcome based on
+     * the response, and then returns a completed future.
      *
      * @param outcome outcome to be populate
      * @param url URL to which to request was sent
      * @param response raw response to process
-     * @return the outcome
+     * @return a future to cancel or await the outcome
      */
-    protected OperationOutcome processResponse(OperationOutcome outcome, String url, Response rawResponse) {
+    protected CompletableFuture<OperationOutcome> processResponse(OperationOutcome outcome, String url,
+                    Response rawResponse) {
 
         logger.info("{}.{}: response received for {}", params.getActor(), params.getOperation(), params.getRequestId());
 
@@ -173,7 +175,6 @@ public abstract class HttpOperation<T> extends OperationPartial {
         T response;
         if (responseClass == String.class) {
             response = responseClass.cast(strResponse);
-
         } else {
             try {
                 response = makeCoder().decode(strResponse, responseClass);
@@ -187,26 +188,40 @@ public abstract class HttpOperation<T> extends OperationPartial {
         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);
+            return CompletableFuture.completedFuture(setOutcome(outcome, PolicyResult.FAILURE, response));
         }
 
         logger.info("{}.{} request succeeded for {}", params.getActor(), params.getOperation(), params.getRequestId());
-        setOutcome(outcome, PolicyResult.SUCCESS);
-        postProcessResponse(outcome, url, rawResponse, response);
+        setOutcome(outcome, PolicyResult.SUCCESS, response);
+        return postProcessResponse(outcome, url, rawResponse, response);
+    }
 
-        return outcome;
+    /**
+     * Sets an operation's outcome and default message based on the result.
+     *
+     * @param outcome operation to be updated
+     * @param result result of the operation
+     * @param response response used to populate the outcome
+     * @return the updated operation
+     */
+    public OperationOutcome setOutcome(OperationOutcome outcome, PolicyResult result, T response) {
+        return setOutcome(outcome, result);
     }
 
     /**
-     * Processes a successful response.
+     * Processes a successful response. This method simply returns the outcome wrapped in
+     * a completed future.
      *
      * @param outcome outcome to be populate
      * @param url URL to which to request was sent
      * @param rawResponse raw response
      * @param response decoded response
+     * @return a future to cancel or await the outcome
      */
-    protected void postProcessResponse(OperationOutcome outcome, String url, Response rawResponse, T response) {
-        // do nothing
+    protected CompletableFuture<OperationOutcome> postProcessResponse(OperationOutcome outcome, String url,
+                    Response rawResponse, T response) {
+
+        return CompletableFuture.completedFuture(outcome);
     }
 
     /**
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
index 0b3497197..680a56f89 100644
--- 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
@@ -68,6 +68,7 @@ import org.slf4j.LoggerFactory;
 public abstract class OperationPartial implements Operation {
     private static final Logger logger = LoggerFactory.getLogger(OperationPartial.class);
     private static final Coder coder = new StandardCoder();
+
     public static final long DEFAULT_RETRY_WAIT_MS = 1000L;
 
     // values extracted from the operator
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/topic/TopicListenerImpl.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/topic/TopicListenerImpl.java
index fcb463518..93beab1cb 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/topic/TopicListenerImpl.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/topic/TopicListenerImpl.java
@@ -45,7 +45,7 @@ import org.slf4j.LoggerFactory;
  */
 public class TopicListenerImpl implements TopicListener {
     private static final Logger logger = LoggerFactory.getLogger(TopicListenerImpl.class);
-    private static StandardCoder coder = new StandardCoder();
+    private static final StandardCoder coder = new StandardCoder();
 
     /**
      * Maps selector to a forwarder.