dcaedt_catalog/commons/src/main/java/org/onap/sdc/dcae/catalog/commons/Futures.java

   1 package org.onap.sdc.dcae.catalog.commons;
   2
   3 import java.util.List;
   4 import java.util.LinkedList;
   5 import java.util.Collections;
   6
   7 import java.util.concurrent.CountDownLatch;
   8 import java.util.function.Function;
   9
  10 import org.onap.sdc.common.onaplog.OnapLoggerDebug;
  11 import org.onap.sdc.common.onaplog.OnapLoggerError;
  12 import org.onap.sdc.dcae.catalog.commons.Future;
  13 import org.onap.sdc.dcae.catalog.commons.FutureHandler;
  14 import org.onap.sdc.common.onaplog.Enums.LogLevel;
  15
  16
  17 /**
  18  */
  19 public class Futures<T> {
  20
  21         private Futures() {
  22         }
  23
  24
  25         public static <T> Future<T>     failedFuture(Throwable theError) {
  26                 return new BasicFuture<T>()
  27                                                         .cause(theError);
  28         }
  29
  30         public static <T> Future<T>     succeededFuture(T theResult) {
  31                 return new BasicFuture<T>()
  32                                                         .result(theResult);
  33         }
  34
  35         public static <T> Future<T>     future() {
  36                 return new BasicFuture<T>();
  37         }
  38
  39         public static <U,V> Future<V> advance(Future<U> theStep,
  40                                                                                                                                                                 final Function<U,V> theResultFunction) {
  41                 return advance(theStep, theResultFunction, Function.identity());
  42         }
  43
  44         public static <U,V> Future<V> advance(Future<U> theStep,
  45                                                                                                                                                                 final Function<U,V> theResultFunction,
  46                                                                                                                                                                 final Function<Throwable, Throwable> theErrorFunction) {
  47                 final Future<V> adv = new BasicFuture<V>();
  48                 theStep.setHandler(new FutureHandler<U>() {
  49                                                                                                                         public void handle(Future<U> theResult) {
  50                                                                                                                                 if (theResult.failed())
  51                                                                                                                                         adv.cause(theErrorFunction.apply(theResult.cause()));
  52                                                                                                                                 else
  53                                                                                                                                         adv.result(theResultFunction.apply(theResult.result()));
  54                                                                                                                         }
  55                                                                                                         });
  56                 return adv;
  57         }
  58
  59         /** */
  60         public static class BasicFuture<T> implements Future<T> {
  61
  62                 protected boolean                succeeded,
  63                                                                                                          failed;
  64
  65                 protected FutureHandler<T> handler;
  66                 protected Throwable                             cause;
  67                 protected T                                                             result;
  68
  69
  70                 protected BasicFuture() {
  71                 }
  72
  73                 public T result() {
  74                         return this.result;
  75                 }
  76
  77                 public Future<T> result(T theResult) {
  78                         this.result = theResult;
  79                         this.succeeded = true;
  80                         this.cause = null;
  81                         this.failed = false;
  82                         callHandler();
  83                         return this;
  84                 }
  85
  86                 public Throwable cause() {
  87                         return this.cause;
  88                 }
  89
  90                 public Future<T> cause(Throwable theCause) {
  91                         this.cause = theCause;
  92                         this.failed = true;
  93                         this.result = null;
  94                         this.succeeded = false;
  95                         callHandler();
  96                         return this;
  97                 }
  98
  99                 public boolean succeeded() {
 100                         return this.succeeded;
 101                 }
 102
 103                 public boolean failed() {
 104                         return this.failed;
 105                 }
 106
 107                 public boolean complete() {
 108                         return this.failed || this.succeeded;
 109                 }
 110
 111                 public Future<T> setHandler(FutureHandler<T> theHandler) {
 112                         this.handler = theHandler;
 113                         callHandler();
 114                         return this;
 115                 }
 116
 117                 public T waitForResult() throws Exception {
 118                         BasicHandler<T> hnd = buildHandler();
 119                         setHandler(hnd);
 120                         hnd.waitForCompletion();
 121                         if (failed())
 122                                 throw (Exception)cause();
 123                         else
 124                                 return result();
 125                 }
 126
 127                 public Future<T> waitForCompletion() throws InterruptedException {
 128                         BasicHandler<T> hnd = buildHandler();
 129                         setHandler(hnd);
 130                         hnd.waitForCompletion();
 131                         return this;
 132                 }
 133
 134                 protected void callHandler() {
 135                         if (this.handler != null && complete()) {
 136                                 this.handler.handle(this);
 137                         }
 138                 }
 139
 140                 protected BasicHandler<T> buildHandler() {
 141                         return new BasicHandler<T>();
 142                 }
 143         }
 144
 145
 146         /** */
 147         public static class BasicHandler<T>
 148                                                                                                 implements FutureHandler<T> {
 149
 150                 protected T                                                     result = null;
 151                 protected Throwable                             error = null;
 152                 protected CountDownLatch        latch = null;
 153
 154                 BasicHandler() {
 155                         this(new CountDownLatch(1));
 156                 }
 157
 158                 BasicHandler(CountDownLatch theLatch) {
 159                         this.latch = theLatch;
 160                 }
 161
 162                 public void handle(Future<T> theResult) {
 163                         process(theResult);
 164                         if (this.latch != null) {
 165                                 this.latch.countDown();
 166                         }
 167                 }
 168
 169                 protected void process(Future<T> theResult) {
 170                         if (theResult.failed()) {
 171                                 this.error = theResult.cause();
 172                         }
 173                         else {
 174                                 this.result = theResult.result();
 175                         }
 176                 }
 177
 178                 public T result(boolean doWait)
 179                                                                                                                         throws InterruptedException, RuntimeException {
 180                         if (doWait) {
 181                                 waitForCompletion();
 182                         }
 183                         if (null == this.error)
 184                                 return this.result;
 185
 186                         throw new RuntimeException(this.error);
 187                 }
 188
 189                 public T result()
 190                                                                                                                         throws InterruptedException, RuntimeException {
 191                         return result(true);
 192                 }
 193
 194                 public BasicHandler<T> waitForCompletion() throws InterruptedException {
 195                         this.latch.await();
 196                         return this;
 197                 }
 198         }
 199
 200         /** */
 201         public static class Accumulator<T>      extends BasicFuture<List<T>>
 202                                                                                                                                                                 implements Future<List<T>> {
 203
 204                 protected List<Future<T>>                       futures = new LinkedList<Future<T>>();
 205                 //protected     List<T>                                                         results = new LinkedList<T>();
 206                 protected BasicHandler<T>                               handler = null;
 207
 208                 private static OnapLoggerError errLogger = OnapLoggerError.getInstance();
 209                 private static OnapLoggerDebug debugLogger = OnapLoggerDebug.getInstance();
 210
 211                 public Accumulator() {
 212                         this.result = new LinkedList<T>();
 213                 }
 214
 215                 public Accumulator<T> add(Future<T> theFuture) {
 216                         debugLogger.log(LogLevel.DEBUG, this.getClass().getName(), "Intersection add");
 217                         this.futures.add(theFuture);
 218                         this.result.add(null);
 219                         return this;
 220                 }
 221
 222                 public Accumulator<T> addAll(Accumulator<T> theFutures) {
 223
 224                         debugLogger.log(LogLevel.DEBUG, this.getClass().getName(), "Intersection addAll");
 225
 226                         return this;
 227                 }
 228
 229                 public Future<List<T>> accumulate() {
 230                         this.futures = Collections.unmodifiableList(this.futures);
 231                         this.handler = new BasicHandler<T>(new CountDownLatch(this.futures.size())) {
 232                                                                                                 protected void process(Future<T> theResult) {
 233                                                                                                         if (theResult.failed()) {
 234                                                                                                                 Accumulator.this.cause = theResult.cause();
 235                                                                                                         }
 236                                                                                                         else {
 237                                                                                                                 Accumulator.this.result.set(
 238                                                                                                                         Accumulator.this.futures.indexOf(theResult), theResult.result());
 239                                                                                                         }
 240                                                                                                         if (this.latch.getCount() == 1) {
 241                                                                                                                 if (Accumulator.this.cause != null)
 242                                                                                                                         Accumulator.this.cause(Accumulator.this.cause);
 243                                                                                                                 else
 244                                                                                                                         Accumulator.this.result(Accumulator.this.result);
 245                                                                                                         }
 246                                                                                                 }
 247                                                                                  };
 248                         futures.stream()
 249                                                         .forEach(f -> f.setHandler(this.handler));
 250
 251                         return this;
 252                 }
 253
 254         }
 255
 256
 257 }