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
9 package org.opendaylight.openflowplugin.impl.util;
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 java.util.concurrent.atomic.AtomicInteger;
20 import javax.annotation.Nullable;
21 import org.apache.commons.lang3.tuple.Pair;
22 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
23 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
24 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
25 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
26 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
27 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
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.slf4j.Logger;
46 import org.slf4j.LoggerFactory;
49 * Created by Martin Bobak <mbobak@cisco.com> on 8.4.2015.
51 public final class FlowUtil {
53 private static final String ALIEN_SYSTEM_FLOW_ID = "#UF$TABLE*";
54 private static final AtomicInteger unaccountedFlowsCounter = new AtomicInteger(0);
55 private static final Logger LOG = LoggerFactory.getLogger(FlowUtil.class);
56 private static final RpcResultBuilder<List<BatchFailedFlowsOutput>> SUCCESSFUL_FLOW_OUTPUT_RPC_RESULT =
57 RpcResultBuilder.success(Collections.<BatchFailedFlowsOutput>emptyList());
59 /** Attach barrier response to given {@link RpcResult}<RemoveFlowsBatchOutput> */
60 public static final Function<Pair<RpcResult<RemoveFlowsBatchOutput>, RpcResult<Void>>, RpcResult<RemoveFlowsBatchOutput>>
61 FLOW_REMOVE_COMPOSING_TRANSFORM = createComposingFunction();
63 /** Attach barrier response to given {@link RpcResult}<AddFlowsBatchOutput> */
64 public static final Function<Pair<RpcResult<AddFlowsBatchOutput>, RpcResult<Void>>, RpcResult<AddFlowsBatchOutput>>
65 FLOW_ADD_COMPOSING_TRANSFORM = createComposingFunction();
67 /** Attach barrier response to given {@link RpcResult}<UpdateFlowsBatchOutput> */
68 public static final Function<Pair<RpcResult<UpdateFlowsBatchOutput>, RpcResult<Void>>, RpcResult<UpdateFlowsBatchOutput>>
69 FLOW_UPDATE_COMPOSING_TRANSFORM = createComposingFunction();
72 * Gather errors into collection and wrap it into {@link RpcResult} and propagate all {@link RpcError}
74 public static final Function<RpcResult<List<BatchFailedFlowsOutput>>, RpcResult<RemoveFlowsBatchOutput>> FLOW_REMOVE_TRANSFORM =
75 new Function<RpcResult<List<BatchFailedFlowsOutput>>, RpcResult<RemoveFlowsBatchOutput>>() {
78 public RpcResult<RemoveFlowsBatchOutput> apply(@Nullable final RpcResult<List<BatchFailedFlowsOutput>> batchFlowsCumulativeResult) {
79 final RemoveFlowsBatchOutput batchOutput = new RemoveFlowsBatchOutputBuilder()
80 .setBatchFailedFlowsOutput(batchFlowsCumulativeResult.getResult()).build();
82 final RpcResultBuilder<RemoveFlowsBatchOutput> resultBld =
83 createCumulativeRpcResult(batchFlowsCumulativeResult, batchOutput);
84 return resultBld.build();
89 * Gather errors into collection and wrap it into {@link RpcResult} and propagate all {@link RpcError}
91 public static final Function<RpcResult<List<BatchFailedFlowsOutput>>, RpcResult<AddFlowsBatchOutput>> FLOW_ADD_TRANSFORM =
92 new Function<RpcResult<List<BatchFailedFlowsOutput>>, RpcResult<AddFlowsBatchOutput>>() {
95 public RpcResult<AddFlowsBatchOutput> apply(@Nullable final RpcResult<List<BatchFailedFlowsOutput>> batchFlowsCumulativeResult) {
96 final AddFlowsBatchOutput batchOutput = new AddFlowsBatchOutputBuilder()
97 .setBatchFailedFlowsOutput(batchFlowsCumulativeResult.getResult()).build();
99 final RpcResultBuilder<AddFlowsBatchOutput> resultBld =
100 createCumulativeRpcResult(batchFlowsCumulativeResult, batchOutput);
101 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>>, RpcResult<UpdateFlowsBatchOutput>> FLOW_UPDATE_TRANSFORM =
109 new Function<RpcResult<List<BatchFailedFlowsOutput>>, RpcResult<UpdateFlowsBatchOutput>>() {
112 public RpcResult<UpdateFlowsBatchOutput> apply(@Nullable final RpcResult<List<BatchFailedFlowsOutput>> batchFlowsCumulativeResult) {
113 final UpdateFlowsBatchOutput batchOutput = new UpdateFlowsBatchOutputBuilder()
114 .setBatchFailedFlowsOutput(batchFlowsCumulativeResult.getResult()).build();
116 final RpcResultBuilder<UpdateFlowsBatchOutput> resultBld =
117 createCumulativeRpcResult(batchFlowsCumulativeResult, batchOutput);
118 return resultBld.build();
123 throw new IllegalStateException("This class should not be instantiated.");
127 * Wrap given list of problematic flow-ids into {@link RpcResult} of given type.
129 * @param batchFlowsCumulativeResult list of ids failed flows
131 * @param <T> 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>
135 RpcResultBuilder<T> createCumulativeRpcResult(final @Nullable 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());
148 public static FlowId createAlienFlowId(final short tableId) {
149 final StringBuilder sBuilder = new StringBuilder(ALIEN_SYSTEM_FLOW_ID)
150 .append(tableId).append('-').append(unaccountedFlowsCounter.incrementAndGet());
151 String alienId = sBuilder.toString();
152 return new FlowId(alienId);
157 * Factory method: create {@link Function} which attaches barrier response to given {@link RpcResult}<T>
158 * and changes success flag if needed.
160 * Original rpcResult is the {@link Pair#getLeft()} and barrier result is the {@link Pair#getRight()}.
162 * @param <T> type of rpcResult value
163 * @return reusable static function
166 static <T extends BatchFlowOutputListGrouping>
167 Function<Pair<RpcResult<T>, RpcResult<Void>>, RpcResult<T>> createComposingFunction() {
168 return new Function<Pair<RpcResult<T>, RpcResult<Void>>, RpcResult<T>>() {
171 public RpcResult<T> apply(@Nullable final Pair<RpcResult<T>, RpcResult<Void>> input) {
172 final RpcResultBuilder<T> resultBld;
173 if (input.getLeft().isSuccessful() && input.getRight().isSuccessful()) {
174 resultBld = RpcResultBuilder.success();
176 resultBld = RpcResultBuilder.failed();
179 final ArrayList<RpcError> rpcErrors = new ArrayList<>(input.getLeft().getErrors());
180 rpcErrors.addAll(input.getRight().getErrors());
181 resultBld.withRpcErrors(rpcErrors);
183 resultBld.withResult(input.getLeft().getResult());
185 return resultBld.build();
191 * @param nodePath path to {@link Node}
192 * @param tableId path to {@link Table} under {@link Node}
193 * @param flowId path to {@link Flow} under {@link Table}
194 * @return instance identifier assembled for given node, table and flow
196 public static FlowRef buildFlowPath(final InstanceIdentifier<Node> nodePath,
197 final short tableId, final FlowId flowId) {
198 final KeyedInstanceIdentifier<Flow, FlowKey> flowPath = nodePath
199 .augmentation(FlowCapableNode.class)
200 .child(Table.class, new TableKey(tableId))
201 .child(Flow.class, new FlowKey(new FlowId(flowId)));
203 return new FlowRef(flowPath);
207 * Factory method: creates {@link Function} which keeps info of original inputs (passed to flow-rpc) and processes
208 * list of all flow-rpc results.
210 * @param inputBatchFlows collection of problematic flow-ids wrapped in container of given type <O>
211 * @param <O> result container type
212 * @return static reusable function
214 public static <O> Function<List<RpcResult<O>>, RpcResult<List<BatchFailedFlowsOutput>>> createCumulatingFunction(
215 final List<? extends BatchFlowIdGrouping> inputBatchFlows) {
216 return new Function<List<RpcResult<O>>, RpcResult<List<BatchFailedFlowsOutput>>>() {
219 public RpcResult<List<BatchFailedFlowsOutput>> apply(@Nullable final List<RpcResult<O>> innerInput) {
220 final int sizeOfFutures = innerInput.size();
221 final int sizeOfInputBatch = inputBatchFlows.size();
222 Preconditions.checkArgument(sizeOfFutures == sizeOfInputBatch,
223 "wrong amount of returned futures: {} <> {}", sizeOfFutures, sizeOfInputBatch);
225 final ArrayList<BatchFailedFlowsOutput> batchFlows = new ArrayList<>(sizeOfFutures);
226 final Iterator<? extends BatchFlowIdGrouping> batchFlowIterator = inputBatchFlows.iterator();
228 Collection<RpcError> flowErrors = new ArrayList<>(sizeOfFutures);
231 for (RpcResult<O> flowModOutput : innerInput) {
232 final FlowId flowId = batchFlowIterator.next().getFlowId();
234 if (!flowModOutput.isSuccessful()) {
235 batchFlows.add(new BatchFailedFlowsOutputBuilder()
237 .setBatchOrder(batchOrder)
239 flowErrors.addAll(flowModOutput.getErrors());
244 final RpcResultBuilder<List<BatchFailedFlowsOutput>> resultBuilder;
245 if (!flowErrors.isEmpty()) {
246 resultBuilder = RpcResultBuilder.<List<BatchFailedFlowsOutput>>failed()
247 .withRpcErrors(flowErrors).withResult(batchFlows);
249 resultBuilder = SUCCESSFUL_FLOW_OUTPUT_RPC_RESULT;
251 return resultBuilder.build();