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