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 package org.opendaylight.openflowplugin.impl.util;
   9
  10 import com.google.common.annotations.VisibleForTesting;
  11 import com.google.common.base.Function;
  12 import com.google.common.base.Preconditions;
  13 import java.util.ArrayList;
  14 import java.util.Collection;
  15 import java.util.Collections;
  16 import java.util.Iterator;
  17 import java.util.List;
  18 import java.util.Map;
  19 import org.apache.commons.lang3.tuple.Pair;
  20 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  21 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
  22 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  23 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
  24 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  25 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
  26 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev150304.SendBarrierOutput;
  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.Identifiable;
  40 import org.opendaylight.yangtools.yang.binding.Identifier;
  41 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  42 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
  43 import org.opendaylight.yangtools.yang.binding.util.BindingMap;
  44 import org.opendaylight.yangtools.yang.common.RpcError;
  45 import org.opendaylight.yangtools.yang.common.RpcResult;
  46 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  47 import org.opendaylight.yangtools.yang.common.Uint16;
  48 import org.opendaylight.yangtools.yang.common.Uint8;
  49
  50 public final class FlowUtil {
  51
  52     private static final RpcResultBuilder<List<BatchFailedFlowsOutput>> SUCCESSFUL_FLOW_OUTPUT_RPC_RESULT =
  53             RpcResultBuilder.success(Collections.emptyList());
  54
  55     /**
  56      * Attach barrier response to given {@link RpcResult}&lt;RemoveFlowsBatchOutput&gt;.
  57      */
  58     public static final Function<Pair<RpcResult<RemoveFlowsBatchOutput>,
  59                                  RpcResult<SendBarrierOutput>>,
  60                                  RpcResult<RemoveFlowsBatchOutput>>
  61             FLOW_REMOVE_COMPOSING_TRANSFORM = createComposingFunction();
  62
  63     /**
  64      * Attach barrier response to given {@link RpcResult}&lt;AddFlowsBatchOutput&gt;.
  65      */
  66     public static final Function<Pair<RpcResult<AddFlowsBatchOutput>, RpcResult<SendBarrierOutput>>,
  67             RpcResult<AddFlowsBatchOutput>>
  68             FLOW_ADD_COMPOSING_TRANSFORM = createComposingFunction();
  69
  70     /**
  71      * Attach barrier response to given {@link RpcResult}&lt;UpdateFlowsBatchOutput&gt;.
  72      */
  73     public static final Function<Pair<RpcResult<UpdateFlowsBatchOutput>,
  74                                  RpcResult<SendBarrierOutput>>,
  75                                  RpcResult<UpdateFlowsBatchOutput>>
  76             FLOW_UPDATE_COMPOSING_TRANSFORM = createComposingFunction();
  77
  78     /**
  79      * Gather errors into collection and wrap it into {@link RpcResult} and propagate all {@link RpcError}.
  80      */
  81     public static final Function<RpcResult<List<BatchFailedFlowsOutput>>,
  82         RpcResult<RemoveFlowsBatchOutput>> FLOW_REMOVE_TRANSFORM =
  83             batchFlowsCumulativeResult -> {
  84                 final RemoveFlowsBatchOutput batchOutput = new RemoveFlowsBatchOutputBuilder()
  85                         .setBatchFailedFlowsOutput(index(batchFlowsCumulativeResult.getResult())).build();
  86
  87                 final RpcResultBuilder<RemoveFlowsBatchOutput> resultBld =
  88                         createCumulativeRpcResult(batchFlowsCumulativeResult, batchOutput);
  89                 return resultBld.build();
  90             };
  91
  92     /**
  93      * Gather errors into collection and wrap it into {@link RpcResult} and propagate all {@link RpcError}.
  94      */
  95     public static final Function<RpcResult<List<BatchFailedFlowsOutput>>,
  96         RpcResult<AddFlowsBatchOutput>> FLOW_ADD_TRANSFORM =
  97             batchFlowsCumulativeResult -> {
  98                 final AddFlowsBatchOutput batchOutput = new AddFlowsBatchOutputBuilder()
  99                         .setBatchFailedFlowsOutput(index(batchFlowsCumulativeResult.getResult())).build();
 100
 101                 final RpcResultBuilder<AddFlowsBatchOutput> resultBld =
 102                         createCumulativeRpcResult(batchFlowsCumulativeResult, batchOutput);
 103                 return resultBld.build();
 104             };
 105
 106     /**
 107      * Gather errors into collection and wrap it into {@link RpcResult} and propagate all {@link RpcError}.
 108      */
 109     public static final Function<RpcResult<List<BatchFailedFlowsOutput>>,
 110         RpcResult<UpdateFlowsBatchOutput>> FLOW_UPDATE_TRANSFORM =
 111             batchFlowsCumulativeResult -> {
 112                 final UpdateFlowsBatchOutput batchOutput = new UpdateFlowsBatchOutputBuilder()
 113                         .setBatchFailedFlowsOutput(index(batchFlowsCumulativeResult.getResult())).build();
 114
 115                 final RpcResultBuilder<UpdateFlowsBatchOutput> resultBld =
 116                         createCumulativeRpcResult(batchFlowsCumulativeResult, batchOutput);
 117                 return resultBld.build();
 118             };
 119
 120     private FlowUtil() {
 121         // Hidden on purpose
 122     }
 123
 124     static <K extends Identifier<V>, V extends Identifiable<K>> Map<K, V> index(final List<V> list) {
 125         return list == null ? null : BindingMap.ordered(list);
 126     }
 127
 128     /**
 129      * Wrap given list of problematic flow-ids into {@link RpcResult} of given type.
 130      *
 131      * @param batchFlowsCumulativeResult list of ids failed flows
 132      * @param batchOutput flow operation type
 133      * @return batch flow operation output of given type containing list of flow-ids and corresponding success flag
 134      */
 135     private static <T extends BatchFlowOutputListGrouping> RpcResultBuilder<T> createCumulativeRpcResult(
 136             final RpcResult<List<BatchFailedFlowsOutput>> batchFlowsCumulativeResult,
 137             final T batchOutput) {
 138         final RpcResultBuilder<T> resultBld;
 139         if (batchFlowsCumulativeResult.isSuccessful()) {
 140             resultBld = RpcResultBuilder.success(batchOutput);
 141         } else {
 142             resultBld = RpcResultBuilder.failed();
 143             resultBld.withResult(batchOutput)
 144                     .withRpcErrors(batchFlowsCumulativeResult.getErrors());
 145         }
 146         return resultBld;
 147     }
 148
 149     /**
 150      * Factory method: create {@link Function} which attaches barrier response to given {@link RpcResult}&lt;T&gt;
 151      * and changes success flag if needed.
 152      * <br>
 153      * Original rpcResult is the {@link Pair#getLeft()} and barrier result is the {@link Pair#getRight()}.
 154      *
 155      * @param <T> type of rpcResult value
 156      * @return reusable static function
 157      */
 158     @VisibleForTesting
 159     static <T extends BatchFlowOutputListGrouping>
 160         Function<Pair<RpcResult<T>, RpcResult<SendBarrierOutput>>, RpcResult<T>> createComposingFunction() {
 161         return input -> {
 162             final RpcResultBuilder<T> resultBld;
 163             if (input.getLeft().isSuccessful() && input.getRight().isSuccessful()) {
 164                 resultBld = RpcResultBuilder.success();
 165             } else {
 166                 resultBld = RpcResultBuilder.failed();
 167             }
 168
 169             final ArrayList<RpcError> rpcErrors = new ArrayList<>(input.getLeft().getErrors());
 170             rpcErrors.addAll(input.getRight().getErrors());
 171             resultBld.withRpcErrors(rpcErrors);
 172
 173             resultBld.withResult(input.getLeft().getResult());
 174
 175             return resultBld.build();
 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 Uint8 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 Collection<? extends BatchFlowIdGrouping> inputBatchFlows) {
 207         return new CumulatingFunction<O>(inputBatchFlows).invoke();
 208     }
 209
 210     private static class CumulatingFunction<O> {
 211         private final Collection<? extends BatchFlowIdGrouping> inputBatchFlows;
 212
 213         CumulatingFunction(final Collection<? extends BatchFlowIdGrouping> inputBatchFlows) {
 214             this.inputBatchFlows = inputBatchFlows;
 215         }
 216
 217         public Function<List<RpcResult<O>>, RpcResult<List<BatchFailedFlowsOutput>>> invoke() {
 218             return (final List<RpcResult<O>> innerInput) -> {
 219                 final int sizeOfFutures = innerInput.size();
 220                 final int sizeOfInputBatch = inputBatchFlows.size();
 221                 Preconditions.checkArgument(sizeOfFutures == sizeOfInputBatch,
 222                         "wrong amount of returned futures: {} <> {}", sizeOfFutures, sizeOfInputBatch);
 223
 224                 final ArrayList<BatchFailedFlowsOutput> batchFlows = new ArrayList<>(sizeOfFutures);
 225                 final Iterator<? extends BatchFlowIdGrouping> batchFlowIterator = inputBatchFlows.iterator();
 226
 227                 Collection<RpcError> flowErrors = new ArrayList<>(sizeOfFutures);
 228
 229                 int batchOrder = 0;
 230                 for (RpcResult<O> flowModOutput : innerInput) {
 231                     final FlowId flowId = batchFlowIterator.next().getFlowId();
 232
 233                     if (!flowModOutput.isSuccessful()) {
 234                         batchFlows.add(new BatchFailedFlowsOutputBuilder()
 235                                 .setFlowId(flowId)
 236                                 .setBatchOrder(Uint16.valueOf(batchOrder))
 237                                 .build());
 238                         flowErrors.addAll(flowModOutput.getErrors());
 239                     }
 240                     batchOrder++;
 241                 }
 242
 243                 final RpcResultBuilder<List<BatchFailedFlowsOutput>> resultBuilder;
 244                 if (!flowErrors.isEmpty()) {
 245                     resultBuilder = RpcResultBuilder.<List<BatchFailedFlowsOutput>>failed()
 246                             .withRpcErrors(flowErrors).withResult(batchFlows);
 247                 } else {
 248                     resultBuilder = SUCCESSFUL_FLOW_OUTPUT_RPC_RESULT;
 249                 }
 250                 return resultBuilder.build();
 251             };
 252         }
 253     }
 254 }