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