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

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * ONAP
   4  * ================================================================================
   5  * Copyright (C) 2020 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.policy.controlloop.actorserviceprovider.impl;
  22
  23 import java.util.ArrayDeque;
  24 import java.util.ArrayList;
  25 import java.util.Arrays;
  26 import java.util.LinkedHashMap;
  27 import java.util.LinkedList;
  28 import java.util.List;
  29 import java.util.Map;
  30 import java.util.Queue;
  31 import java.util.concurrent.CompletableFuture;
  32 import java.util.concurrent.CompletionException;
  33 import java.util.concurrent.Executor;
  34 import java.util.concurrent.TimeUnit;
  35 import java.util.concurrent.TimeoutException;
  36 import java.util.function.BiConsumer;
  37 import java.util.function.Function;
  38 import java.util.function.Supplier;
  39 import java.util.function.UnaryOperator;
  40 import lombok.Getter;
  41 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
  42 import org.onap.policy.common.endpoints.utils.NetLoggerUtil;
  43 import org.onap.policy.common.endpoints.utils.NetLoggerUtil.EventType;
  44 import org.onap.policy.common.utils.coder.Coder;
  45 import org.onap.policy.common.utils.coder.CoderException;
  46 import org.onap.policy.common.utils.coder.StandardCoder;
  47 import org.onap.policy.controlloop.ControlLoopOperation;
  48 import org.onap.policy.controlloop.actorserviceprovider.CallbackManager;
  49 import org.onap.policy.controlloop.actorserviceprovider.Operation;
  50 import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
  51 import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
  52 import org.onap.policy.controlloop.actorserviceprovider.parameters.OperatorConfig;
  53 import org.onap.policy.controlloop.actorserviceprovider.pipeline.PipelineControllerFuture;
  54 import org.onap.policy.controlloop.policy.PolicyResult;
  55 import org.slf4j.Logger;
  56 import org.slf4j.LoggerFactory;
  57
  58 /**
  59  * Partial implementation of an operator. In general, it's preferable that subclasses
  60  * would override {@link #startOperationAsync(int, OperationOutcome)
  61  * startOperationAsync()}. However, if that proves to be too difficult, then they can
  62  * simply override {@link #doOperation(int, OperationOutcome) doOperation()}. In addition,
  63  * if the operation requires any preprocessor steps, the subclass may choose to override
  64  * {@link #startPreprocessorAsync()}.
  65  * <p/>
  66  * The futures returned by the methods within this class can be canceled, and will
  67  * propagate the cancellation to any subtasks. Thus it is also expected that any futures
  68  * returned by overridden methods will do the same. Of course, if a class overrides
  69  * {@link #doOperation(int, OperationOutcome) doOperation()}, then there's little that can
  70  * be done to cancel that particular operation.
  71  * <p/>
  72  * In general tasks in a pipeline are executed by the same thread. However, the following
  73  * should always be executed via the executor specified in "params":
  74  * <ul>
  75  * <li>start callback</li>
  76  * <li>completion callback</li>
  77  * <li>controller completion (i.e., delayedComplete())</li>
  78  * </ul>
  79  */
  80 public abstract class OperationPartial implements Operation {
  81     private static final Logger logger = LoggerFactory.getLogger(OperationPartial.class);
  82     private static final Coder coder = new StandardCoder();
  83
  84     public static final String GUARD_ACTOR_NAME = "GUARD";
  85     public static final String GUARD_OPERATION_NAME = "Decision";
  86     public static final long DEFAULT_RETRY_WAIT_MS = 1000L;
  87
  88     private final OperatorConfig config;
  89
  90     /**
  91      * Operation parameters.
  92      */
  93     protected final ControlLoopOperationParams params;
  94
  95     @Getter
  96     private final String fullName;
  97
  98
  99     /**
 100      * Constructs the object.
 101      *
 102      * @param params operation parameters
 103      * @param config configuration for this operation
 104      */
 105     public OperationPartial(ControlLoopOperationParams params, OperatorConfig config) {
 106         this.params = params;
 107         this.config = config;
 108         this.fullName = params.getActor() + "." + params.getOperation();
 109     }
 110
 111     public Executor getBlockingExecutor() {
 112         return config.getBlockingExecutor();
 113     }
 114
 115     public String getActorName() {
 116         return params.getActor();
 117     }
 118
 119     public String getName() {
 120         return params.getOperation();
 121     }
 122
 123     @Override
 124     public CompletableFuture<OperationOutcome> start() {
 125         // allocate a controller for the entire operation
 126         final PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
 127
 128         CompletableFuture<OperationOutcome> preproc = startPreprocessorAsync();
 129         if (preproc == null) {
 130             // no preprocessor required - just start the operation
 131             return startOperationAttempt(controller, 1);
 132         }
 133
 134         /*
 135          * Do preprocessor first and then, if successful, start the operation. Note:
 136          * operations create their own outcome, ignoring the outcome from any previous
 137          * steps.
 138          *
 139          * Wrap the preprocessor to ensure "stop" is propagated to it.
 140          */
 141         // @formatter:off
 142         controller.wrap(preproc)
 143                         .exceptionally(fromException("preprocessor of operation"))
 144                         .thenCompose(handlePreprocessorFailure(controller))
 145                         .thenCompose(unusedOutcome -> startOperationAttempt(controller, 1))
 146                         .whenCompleteAsync(controller.delayedComplete(), params.getExecutor());
 147         // @formatter:on
 148
 149         return controller;
 150     }
 151
 152     /**
 153      * Handles a failure in the preprocessor pipeline. If a failure occurred, then it
 154      * invokes the call-backs, marks the controller complete, and returns an incomplete
 155      * future, effectively halting the pipeline. Otherwise, it returns the outcome that it
 156      * received.
 157      * <p/>
 158      * Assumes that no callbacks have been invoked yet.
 159      *
 160      * @param controller pipeline controller
 161      * @return a function that checks the outcome status and continues, if successful, or
 162      *         indicates a failure otherwise
 163      */
 164     private Function<OperationOutcome, CompletableFuture<OperationOutcome>> handlePreprocessorFailure(
 165                     PipelineControllerFuture<OperationOutcome> controller) {
 166
 167         return outcome -> {
 168
 169             if (outcome != null && isSuccess(outcome)) {
 170                 logger.info("{}: preprocessor succeeded for {}", getFullName(), params.getRequestId());
 171                 return CompletableFuture.completedFuture(outcome);
 172             }
 173
 174             logger.warn("preprocessor failed, discontinuing operation {} for {}", getFullName(), params.getRequestId());
 175
 176             final Executor executor = params.getExecutor();
 177             final CallbackManager callbacks = new CallbackManager();
 178
 179             // propagate "stop" to the callbacks
 180             controller.add(callbacks);
 181
 182             final OperationOutcome outcome2 = params.makeOutcome();
 183
 184             // TODO need a FAILURE_MISSING_DATA (e.g., A&AI)
 185
 186             outcome2.setFinalOutcome(true);
 187             outcome2.setResult(PolicyResult.FAILURE_GUARD);
 188             outcome2.setMessage(outcome != null ? outcome.getMessage() : null);
 189
 190             // @formatter:off
 191             CompletableFuture.completedFuture(outcome2)
 192                             .whenCompleteAsync(callbackStarted(callbacks), executor)
 193                             .whenCompleteAsync(callbackCompleted(callbacks), executor)
 194                             .whenCompleteAsync(controller.delayedComplete(), executor);
 195             // @formatter:on
 196
 197             return new CompletableFuture<>();
 198         };
 199     }
 200
 201     /**
 202      * Invokes the operation's preprocessor step(s) as a "future". This method simply
 203      * returns {@code null}.
 204      * <p/>
 205      * This method assumes the following:
 206      * <ul>
 207      * <li>the operator is alive</li>
 208      * <li>exceptions generated within the pipeline will be handled by the invoker</li>
 209      * </ul>
 210      *
 211      * @return a function that will start the preprocessor and returns its outcome, or
 212      *         {@code null} if this operation needs no preprocessor
 213      */
 214     protected CompletableFuture<OperationOutcome> startPreprocessorAsync() {
 215         return null;
 216     }
 217
 218     /**
 219      * Invokes the operation's guard step(s) as a "future".
 220      * <p/>
 221      * This method assumes the following:
 222      * <ul>
 223      * <li>the operator is alive</li>
 224      * <li>exceptions generated within the pipeline will be handled by the invoker</li>
 225      * </ul>
 226      *
 227      * @return a function that will start the guard checks and returns its outcome, or
 228      *         {@code null} if this operation has no guard
 229      */
 230     protected CompletableFuture<OperationOutcome> startGuardAsync() {
 231         // get the guard payload
 232         Map<String, Object> guardPayload = makeGuardPayload();
 233
 234         // wrap it in a "resource"
 235         Map<String, Object> resource = new LinkedHashMap<>();
 236         resource.put("guard", guardPayload);
 237
 238         Map<String, Object> payload = new LinkedHashMap<>();
 239         payload.put("resource", resource);
 240
 241         /*
 242          * Note: can't use constants from actor.guard, because that would create a
 243          * circular dependency.
 244          */
 245         return params.toBuilder().actor(GUARD_ACTOR_NAME).operation(GUARD_OPERATION_NAME).retry(null).timeoutSec(null)
 246                         .payload(payload).build().start();
 247     }
 248
 249     /**
 250      * Creates a payload to execute a guard operation.
 251      *
 252      * @return a new guard payload
 253      */
 254     protected Map<String, Object> makeGuardPayload() {
 255         Map<String, Object> guard = new LinkedHashMap<>();
 256         guard.put("actor", params.getActor());
 257         guard.put("recipe", params.getOperation());
 258         guard.put("target", params.getTargetEntity());
 259         guard.put("requestId", params.getRequestId());
 260
 261         String clname = params.getContext().getEvent().getClosedLoopControlName();
 262         if (clname != null) {
 263             guard.put("clname", clname);
 264         }
 265
 266         return guard;
 267     }
 268
 269     /**
 270      * Starts the operation attempt, with no preprocessor. When all retries complete, it
 271      * will complete the controller.
 272      *
 273      * @param controller controller for all operation attempts
 274      * @param attempt attempt number, typically starting with 1
 275      * @return a future that will return the final result of all attempts
 276      */
 277     private CompletableFuture<OperationOutcome> startOperationAttempt(
 278                     PipelineControllerFuture<OperationOutcome> controller, int attempt) {
 279
 280         // propagate "stop" to the operation attempt
 281         controller.wrap(startAttemptWithoutRetries(attempt)).thenCompose(retryOnFailure(controller, attempt))
 282                         .whenCompleteAsync(controller.delayedComplete(), params.getExecutor());
 283
 284         return controller;
 285     }
 286
 287     /**
 288      * Starts the operation attempt, without doing any retries.
 289      *
 290      * @param params operation parameters
 291      * @param attempt attempt number, typically starting with 1
 292      * @return a future that will return the result of a single operation attempt
 293      */
 294     private CompletableFuture<OperationOutcome> startAttemptWithoutRetries(int attempt) {
 295
 296         logger.info("{}: start operation attempt {} for {}", getFullName(), attempt, params.getRequestId());
 297
 298         final Executor executor = params.getExecutor();
 299         final OperationOutcome outcome = params.makeOutcome();
 300         final CallbackManager callbacks = new CallbackManager();
 301
 302         // this operation attempt gets its own controller
 303         final PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
 304
 305         // propagate "stop" to the callbacks
 306         controller.add(callbacks);
 307
 308         // @formatter:off
 309         CompletableFuture<OperationOutcome> future = CompletableFuture.completedFuture(outcome)
 310                         .whenCompleteAsync(callbackStarted(callbacks), executor)
 311                         .thenCompose(controller.wrap(outcome2 -> startOperationAsync(attempt, outcome2)));
 312         // @formatter:on
 313
 314         // handle timeouts, if specified
 315         long timeoutMillis = getTimeoutMs(params.getTimeoutSec());
 316         if (timeoutMillis > 0) {
 317             logger.info("{}: set timeout to {}ms for {}", getFullName(), timeoutMillis, params.getRequestId());
 318             future = future.orTimeout(timeoutMillis, TimeUnit.MILLISECONDS);
 319         }
 320
 321         /*
 322          * Note: we re-invoke callbackStarted() just to be sure the callback is invoked
 323          * before callbackCompleted() is invoked.
 324          *
 325          * Note: no need to remove "callbacks" from the pipeline, as we're going to stop
 326          * the pipeline as the last step anyway.
 327          */
 328
 329         // @formatter:off
 330         future.exceptionally(fromException("operation"))
 331                     .thenApply(setRetryFlag(attempt))
 332                     .whenCompleteAsync(callbackStarted(callbacks), executor)
 333                     .whenCompleteAsync(callbackCompleted(callbacks), executor)
 334                     .whenCompleteAsync(controller.delayedComplete(), executor);
 335         // @formatter:on
 336
 337         return controller;
 338     }
 339
 340     /**
 341      * Determines if the outcome was successful.
 342      *
 343      * @param outcome outcome to examine
 344      * @return {@code true} if the outcome was successful
 345      */
 346     protected boolean isSuccess(OperationOutcome outcome) {
 347         return (outcome.getResult() == PolicyResult.SUCCESS);
 348     }
 349
 350     /**
 351      * Determines if the outcome was a failure for this operator.
 352      *
 353      * @param outcome outcome to examine, or {@code null}
 354      * @return {@code true} if the outcome is not {@code null} and was a failure
 355      *         <i>and</i> was associated with this operator, {@code false} otherwise
 356      */
 357     protected boolean isActorFailed(OperationOutcome outcome) {
 358         return (isSameOperation(outcome) && outcome.getResult() == PolicyResult.FAILURE);
 359     }
 360
 361     /**
 362      * Determines if the given outcome is for this operation.
 363      *
 364      * @param outcome outcome to examine
 365      * @return {@code true} if the outcome is for this operation, {@code false} otherwise
 366      */
 367     protected boolean isSameOperation(OperationOutcome outcome) {
 368         return OperationOutcome.isFor(outcome, getActorName(), getName());
 369     }
 370
 371     /**
 372      * Invokes the operation as a "future". This method simply invokes
 373      * {@link #doOperation()} using the {@link #blockingExecutor "blocking executor"},
 374      * returning the result via a "future".
 375      * <p/>
 376      * Note: if the operation uses blocking I/O, then it should <i>not</i> be run using
 377      * the executor in the "params", as that may bring the background thread pool to a
 378      * grinding halt. The {@link #blockingExecutor "blocking executor"} should be used
 379      * instead.
 380      * <p/>
 381      * This method assumes the following:
 382      * <ul>
 383      * <li>the operator is alive</li>
 384      * <li>verifyRunning() has been invoked</li>
 385      * <li>callbackStarted() has been invoked</li>
 386      * <li>the invoker will perform appropriate timeout checks</li>
 387      * <li>exceptions generated within the pipeline will be handled by the invoker</li>
 388      * </ul>
 389      *
 390      * @param attempt attempt number, typically starting with 1
 391      * @return a function that will start the operation and return its result when
 392      *         complete
 393      */
 394     protected CompletableFuture<OperationOutcome> startOperationAsync(int attempt, OperationOutcome outcome) {
 395
 396         return CompletableFuture.supplyAsync(() -> doOperation(attempt, outcome), getBlockingExecutor());
 397     }
 398
 399     /**
 400      * Low-level method that performs the operation. This can make the same assumptions
 401      * that are made by {@link #doOperationAsFuture()}. This particular method simply
 402      * throws an {@link UnsupportedOperationException}.
 403      *
 404      * @param attempt attempt number, typically starting with 1
 405      * @param operation the operation being performed
 406      * @return the outcome of the operation
 407      */
 408     protected OperationOutcome doOperation(int attempt, OperationOutcome operation) {
 409
 410         throw new UnsupportedOperationException("start operation " + getFullName());
 411     }
 412
 413     /**
 414      * Sets the outcome status to FAILURE_RETRIES, if the current operation outcome is
 415      * FAILURE, assuming the policy specifies retries and the retry count has been
 416      * exhausted.
 417      *
 418      * @param attempt latest attempt number, starting with 1
 419      * @return a function to get the next future to execute
 420      */
 421     private Function<OperationOutcome, OperationOutcome> setRetryFlag(int attempt) {
 422
 423         return origOutcome -> {
 424             // ensure we have a non-null outcome
 425             OperationOutcome outcome;
 426             if (origOutcome != null) {
 427                 outcome = origOutcome;
 428             } else {
 429                 logger.warn("{}: null outcome; treating as a failure for {}", getFullName(), params.getRequestId());
 430                 outcome = this.setOutcome(params.makeOutcome(), PolicyResult.FAILURE);
 431             }
 432
 433             // ensure correct actor/operation
 434             outcome.setActor(getActorName());
 435             outcome.setOperation(getName());
 436
 437             // determine if we should retry, based on the result
 438             if (outcome.getResult() != PolicyResult.FAILURE) {
 439                 // do not retry success or other failure types (e.g., exception)
 440                 outcome.setFinalOutcome(true);
 441                 return outcome;
 442             }
 443
 444             int retry = getRetry(params.getRetry());
 445             if (retry <= 0) {
 446                 // no retries were specified
 447                 outcome.setFinalOutcome(true);
 448
 449             } else if (attempt <= retry) {
 450                 // have more retries - not the final outcome
 451                 outcome.setFinalOutcome(false);
 452
 453             } else {
 454                 /*
 455                  * retries were specified and we've already tried them all - change to
 456                  * FAILURE_RETRIES
 457                  */
 458                 logger.info("operation {} retries exhausted for {}", getFullName(), params.getRequestId());
 459                 outcome.setResult(PolicyResult.FAILURE_RETRIES);
 460                 outcome.setFinalOutcome(true);
 461             }
 462
 463             return outcome;
 464         };
 465     }
 466
 467     /**
 468      * Restarts the operation if it was a FAILURE. Assumes that {@link #setRetryFlag(int)}
 469      * was previously invoked, and thus that the "operation" is not {@code null}.
 470      *
 471      * @param controller controller for all of the retries
 472      * @param attempt latest attempt number, starting with 1
 473      * @return a function to get the next future to execute
 474      */
 475     private Function<OperationOutcome, CompletableFuture<OperationOutcome>> retryOnFailure(
 476                     PipelineControllerFuture<OperationOutcome> controller, int attempt) {
 477
 478         return operation -> {
 479             if (!isActorFailed(operation)) {
 480                 // wrong type or wrong operation - just leave it as is
 481                 logger.info("not retrying operation {} for {}", getFullName(), params.getRequestId());
 482                 controller.complete(operation);
 483                 return new CompletableFuture<>();
 484             }
 485
 486             if (getRetry(params.getRetry()) <= 0) {
 487                 // no retries - already marked as FAILURE, so just return it
 488                 logger.info("operation {} no retries for {}", getFullName(), params.getRequestId());
 489                 controller.complete(operation);
 490                 return new CompletableFuture<>();
 491             }
 492
 493             /*
 494              * Retry the operation.
 495              */
 496             long waitMs = getRetryWaitMs();
 497             logger.info("retry operation {} in {}ms for {}", getFullName(), waitMs, params.getRequestId());
 498
 499             return sleep(waitMs, TimeUnit.MILLISECONDS)
 500                             .thenCompose(unused -> startOperationAttempt(controller, attempt + 1));
 501         };
 502     }
 503
 504     /**
 505      * Convenience method that starts a sleep(), running via a future.
 506      *
 507      * @param sleepTime time to sleep
 508      * @param unit time unit
 509      * @return a future that will complete when the sleep completes
 510      */
 511     protected CompletableFuture<Void> sleep(long sleepTime, TimeUnit unit) {
 512         if (sleepTime <= 0) {
 513             return CompletableFuture.completedFuture(null);
 514         }
 515
 516         return new CompletableFuture<Void>().completeOnTimeout(null, sleepTime, unit);
 517     }
 518
 519     /**
 520      * Converts an exception into an operation outcome, returning a copy of the outcome to
 521      * prevent background jobs from changing it.
 522      *
 523      * @param type type of item throwing the exception
 524      * @return a function that will convert an exception into an operation outcome
 525      */
 526     private Function<Throwable, OperationOutcome> fromException(String type) {
 527
 528         return thrown -> {
 529             OperationOutcome outcome = params.makeOutcome();
 530
 531             logger.warn("exception throw by {} {}.{} for {}", type, outcome.getActor(), outcome.getOperation(),
 532                             params.getRequestId(), thrown);
 533
 534             return setOutcome(outcome, thrown);
 535         };
 536     }
 537
 538     /**
 539      * Similar to {@link CompletableFuture#anyOf(CompletableFuture...)}, but it cancels
 540      * any outstanding futures when one completes.
 541      *
 542      * @param futureMakers function to make a future. If the function returns
 543      *        {@code null}, then no future is created for that function. On the other
 544      *        hand, if the function throws an exception, then the previously created
 545      *        functions are canceled and the exception is re-thrown
 546      * @return a future to cancel or await an outcome, or {@code null} if no futures were
 547      *         created. If this future is canceled, then all of the futures will be
 548      *         canceled
 549      */
 550     public CompletableFuture<OperationOutcome> anyOf(
 551                     @SuppressWarnings("unchecked") Supplier<CompletableFuture<OperationOutcome>>... futureMakers) {
 552
 553         return anyOf(Arrays.asList(futureMakers));
 554     }
 555
 556     /**
 557      * Similar to {@link CompletableFuture#anyOf(CompletableFuture...)}, but it cancels
 558      * any outstanding futures when one completes.
 559      *
 560      * @param futureMakers function to make a future. If the function returns
 561      *        {@code null}, then no future is created for that function. On the other
 562      *        hand, if the function throws an exception, then the previously created
 563      *        functions are canceled and the exception is re-thrown
 564      * @return a future to cancel or await an outcome, or {@code null} if no futures were
 565      *         created. If this future is canceled, then all of the futures will be
 566      *         canceled. Similarly, when this future completes, any incomplete futures
 567      *         will be canceled
 568      */
 569     public CompletableFuture<OperationOutcome> anyOf(List<Supplier<CompletableFuture<OperationOutcome>>> futureMakers) {
 570
 571         PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
 572
 573         CompletableFuture<OperationOutcome>[] futures =
 574                         attachFutures(controller, futureMakers, UnaryOperator.identity());
 575
 576         if (futures.length == 0) {
 577             // no futures were started
 578             return null;
 579         }
 580
 581         if (futures.length == 1) {
 582             return futures[0];
 583         }
 584
 585         CompletableFuture.anyOf(futures).thenApply(outcome -> (OperationOutcome) outcome)
 586                         .whenCompleteAsync(controller.delayedComplete(), params.getExecutor());
 587
 588         return controller;
 589     }
 590
 591     /**
 592      * Similar to {@link CompletableFuture#allOf(CompletableFuture...)}.
 593      *
 594      * @param futureMakers function to make a future. If the function returns
 595      *        {@code null}, then no future is created for that function. On the other
 596      *        hand, if the function throws an exception, then the previously created
 597      *        functions are canceled and the exception is re-thrown
 598      * @return a future to cancel or await an outcome, or {@code null} if no futures were
 599      *         created. If this future is canceled, then all of the futures will be
 600      *         canceled
 601      */
 602     public CompletableFuture<OperationOutcome> allOf(
 603                     @SuppressWarnings("unchecked") Supplier<CompletableFuture<OperationOutcome>>... futureMakers) {
 604
 605         return allOf(Arrays.asList(futureMakers));
 606     }
 607
 608     /**
 609      * Similar to {@link CompletableFuture#allOf(CompletableFuture...)}.
 610      *
 611      * @param futureMakers function to make a future. If the function returns
 612      *        {@code null}, then no future is created for that function. On the other
 613      *        hand, if the function throws an exception, then the previously created
 614      *        functions are canceled and the exception is re-thrown
 615      * @return a future to cancel or await an outcome, or {@code null} if no futures were
 616      *         created. If this future is canceled, then all of the futures will be
 617      *         canceled. Similarly, when this future completes, any incomplete futures
 618      *         will be canceled
 619      */
 620     public CompletableFuture<OperationOutcome> allOf(List<Supplier<CompletableFuture<OperationOutcome>>> futureMakers) {
 621         PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
 622
 623         Queue<OperationOutcome> outcomes = new LinkedList<>();
 624
 625         CompletableFuture<OperationOutcome>[] futures =
 626                         attachFutures(controller, futureMakers, future -> future.thenApply(outcome -> {
 627                             synchronized (outcomes) {
 628                                 outcomes.add(outcome);
 629                             }
 630                             return outcome;
 631                         }));
 632
 633         if (futures.length == 0) {
 634             // no futures were started
 635             return null;
 636         }
 637
 638         if (futures.length == 1) {
 639             return futures[0];
 640         }
 641
 642         // @formatter:off
 643         CompletableFuture.allOf(futures)
 644                         .thenApply(unused -> combineOutcomes(outcomes))
 645                         .whenCompleteAsync(controller.delayedComplete(), params.getExecutor());
 646         // @formatter:on
 647
 648         return controller;
 649     }
 650
 651     /**
 652      * Invokes the functions to create the futures and attaches them to the controller.
 653      *
 654      * @param controller master controller for all of the futures
 655      * @param futureMakers futures to be attached to the controller
 656      * @param adorn function that "adorns" the future, possible adding onto its pipeline.
 657      *        Returns the adorned future
 658      * @return an array of futures, possibly zero-length. If the array is of size one,
 659      *         then that one item should be returned instead of the controller
 660      */
 661     private CompletableFuture<OperationOutcome>[] attachFutures(PipelineControllerFuture<OperationOutcome> controller,
 662                     List<Supplier<CompletableFuture<OperationOutcome>>> futureMakers,
 663                     UnaryOperator<CompletableFuture<OperationOutcome>> adorn) {
 664
 665         if (futureMakers.isEmpty()) {
 666             @SuppressWarnings("unchecked")
 667             CompletableFuture<OperationOutcome>[] result = new CompletableFuture[0];
 668             return result;
 669         }
 670
 671         // the last, unadorned future that is created
 672         CompletableFuture<OperationOutcome> lastFuture = null;
 673
 674         List<CompletableFuture<OperationOutcome>> futures = new ArrayList<>(futureMakers.size());
 675
 676         // make each future
 677         for (var maker : futureMakers) {
 678             try {
 679                 CompletableFuture<OperationOutcome> future = maker.get();
 680                 if (future == null) {
 681                     continue;
 682                 }
 683
 684                 // propagate "stop" to the future
 685                 controller.add(future);
 686
 687                 futures.add(adorn.apply(future));
 688
 689                 lastFuture = future;
 690
 691             } catch (RuntimeException e) {
 692                 logger.warn("{}: exception creating 'future' for {}", getFullName(), params.getRequestId());
 693                 controller.cancel(false);
 694                 throw e;
 695             }
 696         }
 697
 698         @SuppressWarnings("unchecked")
 699         CompletableFuture<OperationOutcome>[] result = new CompletableFuture[futures.size()];
 700
 701         if (result.length == 1) {
 702             // special case - return the unadorned future
 703             result[0] = lastFuture;
 704             return result;
 705         }
 706
 707         return futures.toArray(result);
 708     }
 709
 710     /**
 711      * Combines the outcomes from a set of tasks.
 712      *
 713      * @param outcomes outcomes to be examined
 714      * @return the combined outcome
 715      */
 716     private OperationOutcome combineOutcomes(Queue<OperationOutcome> outcomes) {
 717
 718         // identify the outcome with the highest priority
 719         OperationOutcome outcome = outcomes.remove();
 720         int priority = detmPriority(outcome);
 721
 722         for (OperationOutcome outcome2 : outcomes) {
 723             int priority2 = detmPriority(outcome2);
 724
 725             if (priority2 > priority) {
 726                 outcome = outcome2;
 727                 priority = priority2;
 728             }
 729         }
 730
 731         logger.info("{}: combined outcome of tasks is {} for {}", getFullName(),
 732                         (outcome == null ? null : outcome.getResult()), params.getRequestId());
 733
 734         return outcome;
 735     }
 736
 737     /**
 738      * Determines the priority of an outcome based on its result.
 739      *
 740      * @param outcome outcome to examine, or {@code null}
 741      * @return the outcome's priority
 742      */
 743     protected int detmPriority(OperationOutcome outcome) {
 744         if (outcome == null || outcome.getResult() == null) {
 745             return 1;
 746         }
 747
 748         switch (outcome.getResult()) {
 749             case SUCCESS:
 750                 return 0;
 751
 752             case FAILURE_GUARD:
 753                 return 2;
 754
 755             case FAILURE_RETRIES:
 756                 return 3;
 757
 758             case FAILURE:
 759                 return 4;
 760
 761             case FAILURE_TIMEOUT:
 762                 return 5;
 763
 764             case FAILURE_EXCEPTION:
 765             default:
 766                 return 6;
 767         }
 768     }
 769
 770     /**
 771      * Performs a sequence of tasks, stopping if a task fails. A given task's future is
 772      * not created until the previous task completes. The pipeline returns the outcome of
 773      * the last task executed.
 774      *
 775      * @param futureMakers functions to make the futures
 776      * @return a future to cancel the sequence or await the outcome
 777      */
 778     public CompletableFuture<OperationOutcome> sequence(
 779                     @SuppressWarnings("unchecked") Supplier<CompletableFuture<OperationOutcome>>... futureMakers) {
 780
 781         return sequence(Arrays.asList(futureMakers));
 782     }
 783
 784     /**
 785      * Performs a sequence of tasks, stopping if a task fails. A given task's future is
 786      * not created until the previous task completes. The pipeline returns the outcome of
 787      * the last task executed.
 788      *
 789      * @param futureMakers functions to make the futures
 790      * @return a future to cancel the sequence or await the outcome, or {@code null} if
 791      *         there were no tasks to perform
 792      */
 793     public CompletableFuture<OperationOutcome> sequence(
 794                     List<Supplier<CompletableFuture<OperationOutcome>>> futureMakers) {
 795
 796         Queue<Supplier<CompletableFuture<OperationOutcome>>> queue = new ArrayDeque<>(futureMakers);
 797
 798         CompletableFuture<OperationOutcome> nextTask = getNextTask(queue);
 799         if (nextTask == null) {
 800             // no tasks
 801             return null;
 802         }
 803
 804         if (queue.isEmpty()) {
 805             // only one task - just return it rather than wrapping it in a controller
 806             return nextTask;
 807         }
 808
 809         /*
 810          * multiple tasks - need a controller to stop whichever task is currently
 811          * executing
 812          */
 813         final PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
 814         final Executor executor = params.getExecutor();
 815
 816         // @formatter:off
 817         controller.wrap(nextTask)
 818                     .thenCompose(nextTaskOnSuccess(controller, queue))
 819                     .whenCompleteAsync(controller.delayedComplete(), executor);
 820         // @formatter:on
 821
 822         return controller;
 823     }
 824
 825     /**
 826      * Executes the next task in the queue, if the previous outcome was successful.
 827      *
 828      * @param controller pipeline controller
 829      * @param taskQueue queue of tasks to be performed
 830      * @return a future to execute the remaining tasks, or the current outcome, if it's a
 831      *         failure, or if there are no more tasks
 832      */
 833     private Function<OperationOutcome, CompletableFuture<OperationOutcome>> nextTaskOnSuccess(
 834                     PipelineControllerFuture<OperationOutcome> controller,
 835                     Queue<Supplier<CompletableFuture<OperationOutcome>>> taskQueue) {
 836
 837         return outcome -> {
 838             if (!isSuccess(outcome)) {
 839                 // return the failure
 840                 return CompletableFuture.completedFuture(outcome);
 841             }
 842
 843             CompletableFuture<OperationOutcome> nextTask = getNextTask(taskQueue);
 844             if (nextTask == null) {
 845                 // no tasks - just return the success
 846                 return CompletableFuture.completedFuture(outcome);
 847             }
 848
 849             // @formatter:off
 850             return controller
 851                         .wrap(nextTask)
 852                         .thenCompose(nextTaskOnSuccess(controller, taskQueue));
 853             // @formatter:on
 854         };
 855     }
 856
 857     /**
 858      * Gets the next task from the queue, skipping those that are {@code null}.
 859      *
 860      * @param taskQueue task queue
 861      * @return the next task, or {@code null} if the queue is now empty
 862      */
 863     private CompletableFuture<OperationOutcome> getNextTask(
 864                     Queue<Supplier<CompletableFuture<OperationOutcome>>> taskQueue) {
 865
 866         Supplier<CompletableFuture<OperationOutcome>> maker;
 867
 868         while ((maker = taskQueue.poll()) != null) {
 869             CompletableFuture<OperationOutcome> future = maker.get();
 870             if (future != null) {
 871                 return future;
 872             }
 873         }
 874
 875         return null;
 876     }
 877
 878     /**
 879      * Sets the start time of the operation and invokes the callback to indicate that the
 880      * operation has started. Does nothing if the pipeline has been stopped.
 881      * <p/>
 882      * This assumes that the "outcome" is not {@code null}.
 883      *
 884      * @param callbacks used to determine if the start callback can be invoked
 885      * @return a function that sets the start time and invokes the callback
 886      */
 887     private BiConsumer<OperationOutcome, Throwable> callbackStarted(CallbackManager callbacks) {
 888
 889         return (outcome, thrown) -> {
 890
 891             if (callbacks.canStart()) {
 892                 outcome.setStart(callbacks.getStartTime());
 893                 outcome.setEnd(null);
 894
 895                 // pass a copy to the callback
 896                 OperationOutcome outcome2 = new OperationOutcome(outcome);
 897                 outcome2.setFinalOutcome(false);
 898                 params.callbackStarted(outcome2);
 899             }
 900         };
 901     }
 902
 903     /**
 904      * Sets the end time of the operation and invokes the callback to indicate that the
 905      * operation has completed. Does nothing if the pipeline has been stopped.
 906      * <p/>
 907      * This assumes that the "outcome" is not {@code null}.
 908      * <p/>
 909      * Note: the start time must be a reference rather than a plain value, because it's
 910      * value must be gotten on-demand, when the returned function is executed at a later
 911      * time.
 912      *
 913      * @param callbacks used to determine if the end callback can be invoked
 914      * @return a function that sets the end time and invokes the callback
 915      */
 916     private BiConsumer<OperationOutcome, Throwable> callbackCompleted(CallbackManager callbacks) {
 917
 918         return (outcome, thrown) -> {
 919             if (callbacks.canEnd()) {
 920                 outcome.setStart(callbacks.getStartTime());
 921                 outcome.setEnd(callbacks.getEndTime());
 922
 923                 // pass a copy to the callback
 924                 params.callbackCompleted(new OperationOutcome(outcome));
 925             }
 926         };
 927     }
 928
 929     /**
 930      * Sets an operation's outcome and message, based on a throwable.
 931      *
 932      * @param operation operation to be updated
 933      * @return the updated operation
 934      */
 935     public OperationOutcome setOutcome(OperationOutcome operation, Throwable thrown) {
 936         PolicyResult result = (isTimeout(thrown) ? PolicyResult.FAILURE_TIMEOUT : PolicyResult.FAILURE_EXCEPTION);
 937         return setOutcome(operation, result);
 938     }
 939
 940     /**
 941      * Sets an operation's outcome and default message based on the result.
 942      *
 943      * @param operation operation to be updated
 944      * @param result result of the operation
 945      * @return the updated operation
 946      */
 947     public OperationOutcome setOutcome(OperationOutcome operation, PolicyResult result) {
 948         logger.trace("{}: set outcome {} for {}", getFullName(), result, params.getRequestId());
 949         operation.setResult(result);
 950         operation.setMessage(result == PolicyResult.SUCCESS ? ControlLoopOperation.SUCCESS_MSG
 951                         : ControlLoopOperation.FAILED_MSG);
 952
 953         return operation;
 954     }
 955
 956     /**
 957      * Determines if a throwable is due to a timeout.
 958      *
 959      * @param thrown throwable of interest
 960      * @return {@code true} if the throwable is due to a timeout, {@code false} otherwise
 961      */
 962     protected boolean isTimeout(Throwable thrown) {
 963         if (thrown instanceof CompletionException) {
 964             thrown = thrown.getCause();
 965         }
 966
 967         return (thrown instanceof TimeoutException);
 968     }
 969
 970     /**
 971      * Logs a response. If the response is not of type, String, then it attempts to
 972      * pretty-print it into JSON before logging.
 973      *
 974      * @param direction IN or OUT
 975      * @param infra communication infrastructure on which it was published
 976      * @param source source name (e.g., the URL or Topic name)
 977      * @param response response to be logged
 978      * @return the JSON text that was logged
 979      */
 980     public <T> String logMessage(EventType direction, CommInfrastructure infra, String source, T response) {
 981         String json;
 982         try {
 983             if (response == null) {
 984                 json = null;
 985             } else if (response instanceof String) {
 986                 json = response.toString();
 987             } else {
 988                 json = makeCoder().encode(response, true);
 989             }
 990
 991         } catch (CoderException e) {
 992             String type = (direction == EventType.IN ? "response" : "request");
 993             logger.warn("cannot pretty-print {}", type, e);
 994             json = response.toString();
 995         }
 996
 997         logger.info("[{}|{}|{}|]{}{}", direction, infra, source, NetLoggerUtil.SYSTEM_LS, json);
 998
 999         return json;
1000     }
1001
1002     // these may be overridden by subclasses or junit tests
1003
1004     /**
1005      * Gets the retry count.
1006      *
1007      * @param retry retry, extracted from the parameters, or {@code null}
1008      * @return the number of retries, or {@code 0} if no retries were specified
1009      */
1010     protected int getRetry(Integer retry) {
1011         return (retry == null ? 0 : retry);
1012     }
1013
1014     /**
1015      * Gets the retry wait, in milliseconds.
1016      *
1017      * @return the retry wait, in milliseconds
1018      */
1019     protected long getRetryWaitMs() {
1020         return DEFAULT_RETRY_WAIT_MS;
1021     }
1022
1023     /**
1024      * Gets the operation timeout.
1025      *
1026      * @param timeoutSec timeout, in seconds, extracted from the parameters, or
1027      *        {@code null}
1028      * @return the operation timeout, in milliseconds, or {@code 0} if no timeout was
1029      *         specified
1030      */
1031     protected long getTimeoutMs(Integer timeoutSec) {
1032         return (timeoutSec == null ? 0 : TimeUnit.MILLISECONDS.convert(timeoutSec, TimeUnit.SECONDS));
1033     }
1034
1035     // these may be overridden by junit tests
1036
1037     protected Coder makeCoder() {
1038         return coder;
1039     }
1040 }