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