openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/util/FlowUtil.java

   1 /*
   2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.openflowplugin.impl.util;
  10
  11 import com.google.common.annotations.VisibleForTesting;
  12 import com.google.common.base.Function;
  13 import com.google.common.base.Preconditions;
  14 import java.util.ArrayList;
  15 import java.util.Collection;
  16 import java.util.Collections;
  17 import java.util.Iterator;
  18 import java.util.List;
  19 import javax.annotation.Nullable;
  20 import org.apache.commons.lang3.tuple.Pair;
  21 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  22 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
  23 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  24 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
  25 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  26 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
  27 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.FlowRef;
  28 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.AddFlowsBatchOutput;
  29 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.AddFlowsBatchOutputBuilder;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.BatchFlowIdGrouping;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.BatchFlowOutputListGrouping;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.RemoveFlowsBatchOutput;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.RemoveFlowsBatchOutputBuilder;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.UpdateFlowsBatchOutput;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.UpdateFlowsBatchOutputBuilder;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.batch.flow.output.list.grouping.BatchFailedFlowsOutput;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.batch.flow.output.list.grouping.BatchFailedFlowsOutputBuilder;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  39 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  40 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
  41 import org.opendaylight.yangtools.yang.common.RpcError;
  42 import org.opendaylight.yangtools.yang.common.RpcResult;
  43 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  44
  45 public final class FlowUtil {
  46
  47     private static final RpcResultBuilder<List<BatchFailedFlowsOutput>> SUCCESSFUL_FLOW_OUTPUT_RPC_RESULT =
  48             RpcResultBuilder.success(Collections.<BatchFailedFlowsOutput>emptyList());
  49
  50     /**
  51      * Attach barrier response to given {@link RpcResult}&lt;RemoveFlowsBatchOutput&gt;
  52      */
  53     public static final Function<Pair<RpcResult<RemoveFlowsBatchOutput>, RpcResult<Void>>, RpcResult<RemoveFlowsBatchOutput>>
  54             FLOW_REMOVE_COMPOSING_TRANSFORM = createComposingFunction();
  55
  56     /**
  57      * Attach barrier response to given {@link RpcResult}&lt;AddFlowsBatchOutput&gt;
  58      */
  59     public static final Function<Pair<RpcResult<AddFlowsBatchOutput>, RpcResult<Void>>, RpcResult<AddFlowsBatchOutput>>
  60             FLOW_ADD_COMPOSING_TRANSFORM = createComposingFunction();
  61
  62     /**
  63      * Attach barrier response to given {@link RpcResult}&lt;UpdateFlowsBatchOutput&gt;
  64      */
  65     public static final Function<Pair<RpcResult<UpdateFlowsBatchOutput>, RpcResult<Void>>, RpcResult<UpdateFlowsBatchOutput>>
  66             FLOW_UPDATE_COMPOSING_TRANSFORM = createComposingFunction();
  67
  68     /**
  69      * Gather errors into collection and wrap it into {@link RpcResult} and propagate all {@link RpcError}
  70      */
  71     public static final Function<RpcResult<List<BatchFailedFlowsOutput>>, RpcResult<RemoveFlowsBatchOutput>> FLOW_REMOVE_TRANSFORM =
  72             new Function<RpcResult<List<BatchFailedFlowsOutput>>, RpcResult<RemoveFlowsBatchOutput>>() {
  73                 @Nullable
  74                 @Override
  75                 public RpcResult<RemoveFlowsBatchOutput> apply(@Nullable final RpcResult<List<BatchFailedFlowsOutput>> batchFlowsCumulativeResult) {
  76                     final RemoveFlowsBatchOutput batchOutput = new RemoveFlowsBatchOutputBuilder()
  77                             .setBatchFailedFlowsOutput(batchFlowsCumulativeResult.getResult()).build();
  78
  79                     final RpcResultBuilder<RemoveFlowsBatchOutput> resultBld =
  80                             createCumulativeRpcResult(batchFlowsCumulativeResult, batchOutput);
  81                     return resultBld.build();
  82                 }
  83             };
  84
  85     /**
  86      * Gather errors into collection and wrap it into {@link RpcResult} and propagate all {@link RpcError}
  87      */
  88     public static final Function<RpcResult<List<BatchFailedFlowsOutput>>, RpcResult<AddFlowsBatchOutput>> FLOW_ADD_TRANSFORM =
  89             new Function<RpcResult<List<BatchFailedFlowsOutput>>, RpcResult<AddFlowsBatchOutput>>() {
  90                 @Nullable
  91                 @Override
  92                 public RpcResult<AddFlowsBatchOutput> apply(@Nullable final RpcResult<List<BatchFailedFlowsOutput>> batchFlowsCumulativeResult) {
  93                     final AddFlowsBatchOutput batchOutput = new AddFlowsBatchOutputBuilder()
  94                             .setBatchFailedFlowsOutput(batchFlowsCumulativeResult.getResult()).build();
  95
  96                     final RpcResultBuilder<AddFlowsBatchOutput> resultBld =
  97                             createCumulativeRpcResult(batchFlowsCumulativeResult, batchOutput);
  98                     return resultBld.build();
  99                 }
 100             };
 101
 102     /**
 103      * Gather errors into collection and wrap it into {@link RpcResult} and propagate all {@link RpcError}
 104      */
 105     public static final Function<RpcResult<List<BatchFailedFlowsOutput>>, RpcResult<UpdateFlowsBatchOutput>> FLOW_UPDATE_TRANSFORM =
 106             new Function<RpcResult<List<BatchFailedFlowsOutput>>, RpcResult<UpdateFlowsBatchOutput>>() {
 107                 @Nullable
 108                 @Override
 109                 public RpcResult<UpdateFlowsBatchOutput> apply(@Nullable final RpcResult<List<BatchFailedFlowsOutput>> batchFlowsCumulativeResult) {
 110                     final UpdateFlowsBatchOutput batchOutput = new UpdateFlowsBatchOutputBuilder()
 111                             .setBatchFailedFlowsOutput(batchFlowsCumulativeResult.getResult()).build();
 112
 113                     final RpcResultBuilder<UpdateFlowsBatchOutput> resultBld =
 114                             createCumulativeRpcResult(batchFlowsCumulativeResult, batchOutput);
 115                     return resultBld.build();
 116                 }
 117             };
 118
 119     private FlowUtil() {
 120         throw new IllegalStateException("This class should not be instantiated.");
 121     }
 122
 123     /**
 124      * Wrap given list of problematic flow-ids into {@link RpcResult} of given type.
 125      *
 126      * @param batchFlowsCumulativeResult list of ids failed flows
 127      * @param batchOutput
 128      * @param <T>                        flow operation type
 129      * @return batch flow operation output of given type containing list of flow-ids and corresponding success flag
 130      */
 131     private static <T extends BatchFlowOutputListGrouping>
 132     RpcResultBuilder<T> createCumulativeRpcResult(final @Nullable RpcResult<List<BatchFailedFlowsOutput>> batchFlowsCumulativeResult,
 133                                                   final T batchOutput) {
 134         final RpcResultBuilder<T> resultBld;
 135         if (batchFlowsCumulativeResult.isSuccessful()) {
 136             resultBld = RpcResultBuilder.success(batchOutput);
 137         } else {
 138             resultBld = RpcResultBuilder.failed();
 139             resultBld.withResult(batchOutput)
 140                     .withRpcErrors(batchFlowsCumulativeResult.getErrors());
 141         }
 142         return resultBld;
 143     }
 144
 145     /**
 146      * Factory method: create {@link Function} which attaches barrier response to given {@link RpcResult}&lt;T&gt;
 147      * and changes success flag if needed.
 148      * <br>
 149      * Original rpcResult is the {@link Pair#getLeft()} and barrier result is the {@link Pair#getRight()}.
 150      *
 151      * @param <T> type of rpcResult value
 152      * @return reusable static function
 153      */
 154     @VisibleForTesting
 155     static <T extends BatchFlowOutputListGrouping>
 156     Function<Pair<RpcResult<T>, RpcResult<Void>>, RpcResult<T>> createComposingFunction() {
 157         return new Function<Pair<RpcResult<T>, RpcResult<Void>>, RpcResult<T>>() {
 158             @Nullable
 159             @Override
 160             public RpcResult<T> apply(@Nullable final Pair<RpcResult<T>, RpcResult<Void>> input) {
 161                 final RpcResultBuilder<T> resultBld;
 162                 if (input.getLeft().isSuccessful() && input.getRight().isSuccessful()) {
 163                     resultBld = RpcResultBuilder.success();
 164                 } else {
 165                     resultBld = RpcResultBuilder.failed();
 166                 }
 167
 168                 final ArrayList<RpcError> rpcErrors = new ArrayList<>(input.getLeft().getErrors());
 169                 rpcErrors.addAll(input.getRight().getErrors());
 170                 resultBld.withRpcErrors(rpcErrors);
 171
 172                 resultBld.withResult(input.getLeft().getResult());
 173
 174                 return resultBld.build();
 175             }
 176         };
 177     }
 178
 179     /**
 180      * Build flow path flow ref.
 181      *
 182      * @param nodePath path to {@link Node}
 183      * @param tableId  path to {@link Table} under {@link Node}
 184      * @param flowId   path to {@link Flow} under {@link Table}
 185      * @return instance identifier assembled for given node, table and flow
 186      */
 187     public static FlowRef buildFlowPath(final InstanceIdentifier<Node> nodePath,
 188                                         final short tableId, final FlowId flowId) {
 189         final KeyedInstanceIdentifier<Flow, FlowKey> flowPath = nodePath
 190                 .augmentation(FlowCapableNode.class)
 191                 .child(Table.class, new TableKey(tableId))
 192                 .child(Flow.class, new FlowKey(new FlowId(flowId)));
 193
 194         return new FlowRef(flowPath);
 195     }
 196
 197     /**
 198      * Factory method: creates {@link Function} which keeps info of original inputs (passed to flow-rpc) and processes
 199      * list of all flow-rpc results.
 200      *
 201      * @param <O>             result container type
 202      * @param inputBatchFlows collection of problematic flow-ids wrapped in container of given type &lt;O&gt;
 203      * @return static reusable function
 204      */
 205     public static <O> Function<List<RpcResult<O>>, RpcResult<List<BatchFailedFlowsOutput>>> createCumulatingFunction(
 206             final List<? extends BatchFlowIdGrouping> inputBatchFlows) {
 207         return new CumulatingFunction<O>(inputBatchFlows).invoke();
 208     }
 209
 210     private static class CumulatingFunction<O> {
 211         private final List<? extends BatchFlowIdGrouping> inputBatchFlows;
 212
 213         public CumulatingFunction(List<? extends BatchFlowIdGrouping> inputBatchFlows) {
 214             this.inputBatchFlows = inputBatchFlows;
 215         }
 216
 217         public Function<List<RpcResult<O>>, RpcResult<List<BatchFailedFlowsOutput>>> invoke() {
 218             return new Function<List<RpcResult<O>>, RpcResult<List<BatchFailedFlowsOutput>>>() {
 219                 @Nullable
 220                 @Override
 221                 public RpcResult<List<BatchFailedFlowsOutput>> apply(@Nullable final List<RpcResult<O>> innerInput) {
 222                     final int sizeOfFutures = innerInput.size();
 223                     final int sizeOfInputBatch = inputBatchFlows.size();
 224                     Preconditions.checkArgument(sizeOfFutures == sizeOfInputBatch,
 225                             "wrong amount of returned futures: {} <> {}", sizeOfFutures, sizeOfInputBatch);
 226
 227                     final ArrayList<BatchFailedFlowsOutput> batchFlows = new ArrayList<>(sizeOfFutures);
 228                     final Iterator<? extends BatchFlowIdGrouping> batchFlowIterator = inputBatchFlows.iterator();
 229
 230                     Collection<RpcError> flowErrors = new ArrayList<>(sizeOfFutures);
 231
 232                     int batchOrder = 0;
 233                     for (RpcResult<O> flowModOutput : innerInput) {
 234                         final FlowId flowId = batchFlowIterator.next().getFlowId();
 235
 236                         if (!flowModOutput.isSuccessful()) {
 237                             batchFlows.add(new BatchFailedFlowsOutputBuilder()
 238                                     .setFlowId(flowId)
 239                                     .setBatchOrder(batchOrder)
 240                                     .build());
 241                             flowErrors.addAll(flowModOutput.getErrors());
 242                         }
 243                         batchOrder++;
 244                     }
 245
 246                     final RpcResultBuilder<List<BatchFailedFlowsOutput>> resultBuilder;
 247                     if (!flowErrors.isEmpty()) {
 248                         resultBuilder = RpcResultBuilder.<List<BatchFailedFlowsOutput>>failed()
 249                                 .withRpcErrors(flowErrors).withResult(batchFlows);
 250                     } else {
 251                         resultBuilder = SUCCESSFUL_FLOW_OUTPUT_RPC_RESULT;
 252                     }
 253                     return resultBuilder.build();
 254                 }
 255             };
 256         }
 257     }
 258 }