--- /dev/null
+/*-
+ * ============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.pipeline;
+
+import static org.onap.policy.controlloop.actorserviceprovider.Util.ident;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+import lombok.NoArgsConstructor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Pipeline controller, used by operations within the pipeline to determine if they should
+ * continue to run. If {@link #cancel(boolean)} is invoked, it automatically stops the
+ * pipeline.
+ */
+@NoArgsConstructor
+public class PipelineControllerFuture<T> extends CompletableFuture<T> {
+
+ private static final Logger logger = LoggerFactory.getLogger(PipelineControllerFuture.class);
+
+ /**
+ * Tracks items added to this controller via one of the <i>add</i> methods.
+ */
+ private final FutureManager futures = new FutureManager();
+
+
+ /**
+ * Cancels and stops the pipeline, in that order.
+ */
+ @Override
+ public boolean cancel(boolean mayInterruptIfRunning) {
+ try {
+ logger.trace("{}: cancel future", ident(this));
+ return super.cancel(mayInterruptIfRunning);
+
+ } finally {
+ futures.stop();
+ }
+ }
+
+ /**
+ * Generates a function that, when invoked, will remove the given future. This is
+ * typically added onto the end of a pipeline via one of the
+ * {@link CompletableFuture#whenComplete(BiConsumer)} methods.
+ *
+ * @return a function that removes the given future
+ */
+ public <F> BiConsumer<T, Throwable> delayedRemove(Future<F> future) {
+ return (value, thrown) -> {
+ logger.trace("{}: remove future {}", ident(this), ident(future));
+ remove(future);
+ };
+ }
+
+ /**
+ * Generates a function that, when invoked, will remove the given listener. This is
+ * typically added onto the end of a pipeline via one of the
+ * {@link CompletableFuture#whenComplete(BiConsumer)} methods.
+ *
+ * @return a function that removes the given listener
+ */
+ public BiConsumer<T, Throwable> delayedRemove(Runnable listener) {
+ return (value, thrown) -> {
+ logger.trace("{}: remove listener {}", ident(this), ident(listener));
+ remove(listener);
+ };
+ }
+
+ /**
+ * Generates a function that, when invoked, will stop all pipeline listeners and
+ * complete this future. This is typically added onto the end of a pipeline via one of
+ * the {@link CompletableFuture#whenComplete(BiConsumer)} methods.
+ *
+ * @return a function that stops all pipeline listeners
+ */
+ public BiConsumer<T, Throwable> delayedComplete() {
+ return (value, thrown) -> {
+ if (thrown == null) {
+ logger.trace("{}: complete and stop future", ident(this));
+ complete(value);
+ } else {
+ logger.trace("{}: complete exceptionally and stop future", ident(this));
+ completeExceptionally(thrown);
+ }
+
+ futures.stop();
+ };
+ }
+
+ /**
+ * Adds a function whose return value is to be canceled when this controller is
+ * stopped. Note: if the controller is already stopped, then the function will
+ * <i>not</i> be executed.
+ *
+ * @param futureMaker function to be invoked in the future
+ */
+ public <F> Function<F, CompletableFuture<F>> add(Function<F, CompletableFuture<F>> futureMaker) {
+
+ return input -> {
+ if (!isRunning()) {
+ logger.trace("{}: discarded new future", ident(this));
+ return new CompletableFuture<>();
+ }
+
+ CompletableFuture<F> future = futureMaker.apply(input);
+ add(future);
+
+ return future;
+ };
+ }
+
+ public <F> void add(Future<F> future) {
+ logger.trace("{}: add future {}", ident(this), ident(future));
+ futures.add(future);
+ }
+
+ public void add(Runnable listener) {
+ logger.trace("{}: add listener {}", ident(this), ident(listener));
+ futures.add(listener);
+ }
+
+ public boolean isRunning() {
+ return futures.isRunning();
+ }
+
+ public <F> void remove(Future<F> future) {
+ logger.trace("{}: remove future {}", ident(this), ident(future));
+ futures.remove(future);
+ }
+
+ public void remove(Runnable listener) {
+ logger.trace("{}: remove listener {}", ident(this), ident(listener));
+ futures.remove(listener);
+ }
+}
Code Review / policy / models.git / blobdiff