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

   1 package org.onap.sdc.dcae.catalog.commons;
   2
   3 import java.util.List;
   4 import java.util.LinkedList;
   5 import java.util.ArrayList;
   6 import java.util.Collections;
   7 import java.util.concurrent.CountDownLatch;
   8
   9 import org.onap.sdc.common.onaplog.OnapLoggerDebug;
  10 import org.onap.sdc.common.onaplog.OnapLoggerError;
  11 import org.onap.sdc.common.onaplog.Enums.LogLevel;
  12 import org.onap.sdc.dcae.catalog.commons.Action;
  13 import org.onap.sdc.dcae.catalog.commons.Future;
  14 import org.onap.sdc.dcae.catalog.commons.FutureHandler;
  15 import org.onap.sdc.dcae.catalog.commons.Futures;
  16
  17 /**
  18  */
  19 public interface Actions {
  20
  21         /** */
  22         public static interface CompoundAction<T> extends Action<List<T>> {
  23
  24                 public CompoundAction<T> addAction(Action<T> theAction);
  25
  26                 public List<Action<T>> actions();
  27
  28                 public Future<List<T>> execute();
  29         }
  30
  31
  32         public static class BasicCompoundAction<T> implements CompoundAction<T> {
  33
  34                 private LinkedList<Action<T>> actions = new LinkedList<Action<T>>();
  35
  36
  37
  38                 public CompoundAction<T> addAction(Action<T> theAction) {
  39                         this.actions.add(theAction);
  40                         return this;
  41                 }
  42
  43                 public List<Action<T>> actions() {
  44                         return this.actions;
  45                 }
  46
  47                 public Future<List<T>> execute() {
  48                         CompoundFuture<T> cf = new CompoundFuture<T>(this.actions.size());
  49                         for (Action a: this.actions)
  50                                 cf.addFuture(a.execute());
  51                         return cf;
  52                 }
  53         }
  54
  55
  56         public static class CompoundFuture<T> extends Futures.BasicFuture<List<T>> {
  57
  58                 private static OnapLoggerError errLogger = OnapLoggerError.getInstance();
  59                 private static OnapLoggerDebug debugLogger = OnapLoggerDebug.getInstance();
  60
  61                 private LinkedList<Future<T>> futures = new LinkedList<Future<T>>();
  62                 private FutureHandler<T>                        hnd;
  63
  64                 CompoundFuture(int theActionCount) {
  65
  66                         hnd = new Futures.BasicHandler<T>(new CountDownLatch(theActionCount)) {
  67
  68                                                         private List<T> results = new ArrayList<T>(Collections.nCopies(theActionCount, null));
  69
  70                                                         protected void process(Future<T> theResult) {
  71                                                                 synchronized(CompoundFuture.this) {
  72                                                                         if (theResult.failed()) {
  73                                                                                 CompoundFuture.this.cause(theResult.cause());
  74                                                                                 //and stop processing of other results
  75                                                                                 this.results = null;
  76                                                                                 //??
  77                                                                         }
  78                                                                         else {
  79                                                                                 if (this.results != null)
  80                                                                                         this.results.set(futures.indexOf(theResult), theResult.result());
  81                                                                                 debugLogger.log(LogLevel.DEBUG, this.getClass().getName(), "Got result for action {}. Count at {}", futures.indexOf(theResult), this.latch.getCount());
  82                                                                         }
  83                                                                         if (this.latch.getCount() == 1) {//this was the last result
  84                                                                                 debugLogger.log(LogLevel.DEBUG, this.getClass().getName(), "Got all results: {}", this.results);
  85                                                                                 CompoundFuture.this.result(this.results);
  86                                                                         }
  87                                                                 }
  88                                                         }
  89                                                 };
  90                 }
  91
  92                 CompoundFuture<T> addFuture(Future<T> theFuture) {
  93                         synchronized(this) {
  94                                 futures.add(theFuture);
  95                                 theFuture.setHandler(this.hnd);
  96                         }
  97                         return this;
  98                 }
  99
 100         }
 101
 102 /*
 103         public static class CompoundFutureHandler<T> implements FutureHandler<T> {
 104
 105                 protected List<T>                               result = null;
 106                 protected List<Throwable>       error = null;
 107                 protected CountDownLatch        latch = null;
 108
 109                 CompoundFutureHandler(int theResultCount) {
 110                         this(new CountDownLatch(theResultCount));
 111                 }
 112
 113                 public void handle(Future<T> theResult) {
 114                         if (this.latch != null) {
 115                                 this.latch.countDown();
 116                         }
 117                 }
 118
 119                 public T result()
 120                                                                                                                         throws InterruptedException, RuntimeException {
 121                         return result(true);
 122                 }
 123
 124                 public BasicHandler<T> waitForCompletion() throws InterruptedException {
 125                         this.latch.await();
 126                         return this;
 127                 }
 128
 129         }
 130 */
 131
 132         public static class Sequence<T> implements Action<List<T>> {
 133
 134                 private static OnapLoggerError errLogger = OnapLoggerError.getInstance();
 135                 private static OnapLoggerDebug debugLogger = OnapLoggerDebug.getInstance();
 136
 137                 private List<Action<T>>         actions = new LinkedList<Action<T>>();
 138                 private int                                                             current = 0;
 139                 private SequenceFuture<T> future = new SequenceFuture<T>();
 140
 141                 public Sequence<T> add(Action<T> theAction) {
 142                         if (this.current > 0)
 143                                 throw new IllegalStateException("In execution");
 144                         this.actions.add(theAction);
 145                         return this;
 146                 }
 147
 148                 /* we allow 'early' access to the future so that a client can pass its reference while
 149                  * it still builds the sequence, for example.
 150                  */
 151                 public Future<List<T>> future() {
 152                         return this.future;
 153                 }
 154
 155                 //need to add protection when for the 'no action' case
 156                 public Future<List<T>> execute() {
 157                         debugLogger.log(LogLevel.DEBUG, this.getClass().getName(), "Starting serialized execution of {}", actions);
 158                         if (hasNext())
 159                                 next().execute().setHandler(future.hnd);
 160                         return this.future;
 161                 }
 162
 163                 protected boolean hasNext() {
 164                         return this.current < actions.size();
 165                 }
 166
 167                 protected Action next() {
 168                         return actions.get(this.current++);
 169                 }
 170
 171                 private class SequenceFuture<T> extends Futures.BasicFuture<List<T>> {
 172
 173                         private List<T>                                  results = new LinkedList<T>();
 174                         private FutureHandler<T> hnd = new Futures.BasicHandler<T>() {
 175
 176                                 protected void process(Future<T> theResult) {
 177
 178                                         if (theResult.failed()) {
 179                                                 SequenceFuture.this.cause(theResult.cause());
 180                                                 //and stop processing of other results
 181                                         }
 182                                         else {
 183                                                 SequenceFuture.this.results.add(theResult.result());
 184                                                 if (Sequence.this.hasNext()) {
 185                                                         Sequence.this.next().execute().setHandler(this);
 186                                                 }
 187                                                 else {
 188                                                         SequenceFuture.this.result(SequenceFuture.this.results);
 189                                                 }
 190                                         }
 191                                 }
 192                         };
 193
 194
 195                 }
 196
 197
 198
 199         }
 200
 201 }