2 * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.openflowplugin.impl.util;
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.Iterator;
16 import java.util.List;
18 import org.apache.commons.lang3.tuple.Pair;
19 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
20 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
21 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
22 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
23 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
24 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
25 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev150304.SendBarrierOutput;
26 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.FlowRef;
27 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.AddFlowsBatchOutput;
28 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.AddFlowsBatchOutputBuilder;
29 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.BatchFlowIdGrouping;
30 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.BatchFlowOutputListGrouping;
31 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.RemoveFlowsBatchOutput;
32 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.RemoveFlowsBatchOutputBuilder;
33 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.UpdateFlowsBatchOutput;
34 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.UpdateFlowsBatchOutputBuilder;
35 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.batch.flow.output.list.grouping.BatchFailedFlowsOutput;
36 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.batch.flow.output.list.grouping.BatchFailedFlowsOutputBuilder;
37 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
38 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
39 import org.opendaylight.yangtools.yang.binding.Key;
40 import org.opendaylight.yangtools.yang.binding.KeyAware;
41 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
42 import org.opendaylight.yangtools.yang.binding.util.BindingMap;
43 import org.opendaylight.yangtools.yang.common.RpcError;
44 import org.opendaylight.yangtools.yang.common.RpcResult;
45 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
46 import org.opendaylight.yangtools.yang.common.Uint16;
47 import org.opendaylight.yangtools.yang.common.Uint8;
49 public final class FlowUtil {
51 private static final RpcResultBuilder<List<BatchFailedFlowsOutput>> SUCCESSFUL_FLOW_OUTPUT_RPC_RESULT =
52 RpcResultBuilder.success(List.of());
55 * Attach barrier response to given {@link RpcResult}<RemoveFlowsBatchOutput>.
57 public static final Function<Pair<RpcResult<RemoveFlowsBatchOutput>,
58 RpcResult<SendBarrierOutput>>,
59 RpcResult<RemoveFlowsBatchOutput>>
60 FLOW_REMOVE_COMPOSING_TRANSFORM = createComposingFunction();
63 * Attach barrier response to given {@link RpcResult}<AddFlowsBatchOutput>.
65 public static final Function<Pair<RpcResult<AddFlowsBatchOutput>, RpcResult<SendBarrierOutput>>,
66 RpcResult<AddFlowsBatchOutput>>
67 FLOW_ADD_COMPOSING_TRANSFORM = createComposingFunction();
70 * Attach barrier response to given {@link RpcResult}<UpdateFlowsBatchOutput>.
72 public static final Function<Pair<RpcResult<UpdateFlowsBatchOutput>,
73 RpcResult<SendBarrierOutput>>,
74 RpcResult<UpdateFlowsBatchOutput>>
75 FLOW_UPDATE_COMPOSING_TRANSFORM = createComposingFunction();
78 * Gather errors into collection and wrap it into {@link RpcResult} and propagate all {@link RpcError}.
80 public static final Function<RpcResult<List<BatchFailedFlowsOutput>>,
81 RpcResult<RemoveFlowsBatchOutput>> FLOW_REMOVE_TRANSFORM =
82 batchFlowsCumulativeResult -> {
83 final RemoveFlowsBatchOutput batchOutput = new RemoveFlowsBatchOutputBuilder()
84 .setBatchFailedFlowsOutput(index(batchFlowsCumulativeResult.getResult())).build();
86 final RpcResultBuilder<RemoveFlowsBatchOutput> resultBld =
87 createCumulativeRpcResult(batchFlowsCumulativeResult, batchOutput);
88 return resultBld.build();
92 * Gather errors into collection and wrap it into {@link RpcResult} and propagate all {@link RpcError}.
94 public static final Function<RpcResult<List<BatchFailedFlowsOutput>>,
95 RpcResult<AddFlowsBatchOutput>> FLOW_ADD_TRANSFORM =
96 batchFlowsCumulativeResult -> {
97 final AddFlowsBatchOutput batchOutput = new AddFlowsBatchOutputBuilder()
98 .setBatchFailedFlowsOutput(index(batchFlowsCumulativeResult.getResult())).build();
100 final RpcResultBuilder<AddFlowsBatchOutput> resultBld =
101 createCumulativeRpcResult(batchFlowsCumulativeResult, batchOutput);
102 return resultBld.build();
106 * Gather errors into collection and wrap it into {@link RpcResult} and propagate all {@link RpcError}.
108 public static final Function<RpcResult<List<BatchFailedFlowsOutput>>,
109 RpcResult<UpdateFlowsBatchOutput>> FLOW_UPDATE_TRANSFORM =
110 batchFlowsCumulativeResult -> {
111 final UpdateFlowsBatchOutput batchOutput = new UpdateFlowsBatchOutputBuilder()
112 .setBatchFailedFlowsOutput(index(batchFlowsCumulativeResult.getResult())).build();
114 final RpcResultBuilder<UpdateFlowsBatchOutput> resultBld =
115 createCumulativeRpcResult(batchFlowsCumulativeResult, batchOutput);
116 return resultBld.build();
123 static <K extends Key<V>, V extends KeyAware<K>> Map<K, V> index(final List<V> list) {
124 return list == null ? null : BindingMap.ordered(list);
128 * Wrap given list of problematic flow-ids into {@link RpcResult} of given type.
130 * @param batchFlowsCumulativeResult list of ids failed flows
131 * @param batchOutput flow operation type
132 * @return batch flow operation output of given type containing list of flow-ids and corresponding success flag
134 private static <T extends BatchFlowOutputListGrouping> RpcResultBuilder<T> createCumulativeRpcResult(
135 final RpcResult<List<BatchFailedFlowsOutput>> batchFlowsCumulativeResult,
136 final T batchOutput) {
137 final RpcResultBuilder<T> resultBld;
138 if (batchFlowsCumulativeResult.isSuccessful()) {
139 resultBld = RpcResultBuilder.success(batchOutput);
141 resultBld = RpcResultBuilder.failed();
142 resultBld.withResult(batchOutput)
143 .withRpcErrors(batchFlowsCumulativeResult.getErrors());
149 * Factory method: create {@link Function} which attaches barrier response to given {@link RpcResult}<T>
150 * and changes success flag if needed.
152 * Original rpcResult is the {@link Pair#getLeft()} and barrier result is the {@link Pair#getRight()}.
154 * @param <T> type of rpcResult value
155 * @return reusable static function
158 static <T extends BatchFlowOutputListGrouping>
159 Function<Pair<RpcResult<T>, RpcResult<SendBarrierOutput>>, RpcResult<T>> createComposingFunction() {
161 final RpcResultBuilder<T> resultBld;
162 if (input.getLeft().isSuccessful() && input.getRight().isSuccessful()) {
163 resultBld = RpcResultBuilder.success();
165 resultBld = RpcResultBuilder.failed();
168 final ArrayList<RpcError> rpcErrors = new ArrayList<>(input.getLeft().getErrors());
169 rpcErrors.addAll(input.getRight().getErrors());
170 resultBld.withRpcErrors(rpcErrors);
172 resultBld.withResult(input.getLeft().getResult());
174 return resultBld.build();
179 * Build flow path flow ref.
181 * @param nodePath path to {@link Node}
182 * @param tableId path to {@link Table} under {@link Node}
183 * @param flowId path to {@link Flow} under {@link Table}
184 * @return instance identifier assembled for given node, table and flow
186 public static FlowRef buildFlowPath(final InstanceIdentifier<Node> nodePath,
187 final Uint8 tableId, final FlowId flowId) {
188 final KeyedInstanceIdentifier<Flow, FlowKey> flowPath = nodePath
189 .augmentation(FlowCapableNode.class)
190 .child(Table.class, new TableKey(tableId))
191 .child(Flow.class, new FlowKey(new FlowId(flowId)));
193 return new FlowRef(flowPath);
197 * Factory method: creates {@link Function} which keeps info of original inputs (passed to flow-rpc) and processes
198 * list of all flow-rpc results.
200 * @param <O> result container type
201 * @param inputBatchFlows collection of problematic flow-ids wrapped in container of given type <O>
202 * @return static reusable function
204 public static <O> Function<List<RpcResult<O>>, RpcResult<List<BatchFailedFlowsOutput>>> createCumulatingFunction(
205 final Collection<? extends BatchFlowIdGrouping> inputBatchFlows) {
206 return new CumulatingFunction<O>(inputBatchFlows).invoke();
209 private static class CumulatingFunction<O> {
210 private final Collection<? extends BatchFlowIdGrouping> inputBatchFlows;
212 CumulatingFunction(final Collection<? extends BatchFlowIdGrouping> inputBatchFlows) {
213 this.inputBatchFlows = inputBatchFlows;
216 public Function<List<RpcResult<O>>, RpcResult<List<BatchFailedFlowsOutput>>> invoke() {
217 return (final List<RpcResult<O>> innerInput) -> {
218 final int sizeOfFutures = innerInput.size();
219 final int sizeOfInputBatch = inputBatchFlows.size();
220 Preconditions.checkArgument(sizeOfFutures == sizeOfInputBatch,
221 "wrong amount of returned futures: {} <> {}", sizeOfFutures, sizeOfInputBatch);
223 final ArrayList<BatchFailedFlowsOutput> batchFlows = new ArrayList<>(sizeOfFutures);
224 final Iterator<? extends BatchFlowIdGrouping> batchFlowIterator = inputBatchFlows.iterator();
226 Collection<RpcError> flowErrors = new ArrayList<>(sizeOfFutures);
229 for (RpcResult<O> flowModOutput : innerInput) {
230 final FlowId flowId = batchFlowIterator.next().getFlowId();
232 if (!flowModOutput.isSuccessful()) {
233 batchFlows.add(new BatchFailedFlowsOutputBuilder()
235 .setBatchOrder(Uint16.valueOf(batchOrder))
237 flowErrors.addAll(flowModOutput.getErrors());
242 final RpcResultBuilder<List<BatchFailedFlowsOutput>> resultBuilder;
243 if (!flowErrors.isEmpty()) {
244 resultBuilder = RpcResultBuilder.<List<BatchFailedFlowsOutput>>failed()
245 .withRpcErrors(flowErrors).withResult(batchFlows);
247 resultBuilder = SUCCESSFUL_FLOW_OUTPUT_RPC_RESULT;
249 return resultBuilder.build();