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.services.sal;
10 import com.google.common.annotations.VisibleForTesting;
11 import com.google.common.util.concurrent.FutureCallback;
12 import com.google.common.util.concurrent.Futures;
13 import com.google.common.util.concurrent.ListenableFuture;
14 import com.google.common.util.concurrent.SettableFuture;
15 import java.util.ArrayList;
16 import java.util.List;
17 import java.util.Objects;
18 import java.util.concurrent.Future;
19 import org.opendaylight.openflowplugin.api.OFConstants;
20 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
21 import org.opendaylight.openflowplugin.api.openflow.device.RequestContextStack;
22 import org.opendaylight.openflowplugin.api.openflow.registry.flow.DeviceFlowRegistry;
23 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowDescriptor;
24 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowRegistryKey;
25 import org.opendaylight.openflowplugin.impl.registry.flow.FlowDescriptorFactory;
26 import org.opendaylight.openflowplugin.impl.registry.flow.FlowRegistryKeyFactory;
27 import org.opendaylight.openflowplugin.impl.services.multilayer.MultiLayerFlowService;
28 import org.opendaylight.openflowplugin.impl.services.singlelayer.SingleLayerFlowService;
29 import org.opendaylight.openflowplugin.impl.util.ErrorUtil;
30 import org.opendaylight.openflowplugin.impl.util.FlowCreatorUtil;
31 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
32 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
33 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
34 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
35 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
36 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
37 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowInput;
38 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowInputBuilder;
39 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowOutput;
40 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.RemoveFlowInput;
41 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.RemoveFlowInputBuilder;
42 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.RemoveFlowOutput;
43 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.SalFlowService;
44 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.UpdateFlowInput;
45 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.UpdateFlowOutput;
46 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.flow.update.OriginalFlow;
47 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.flow.update.UpdatedFlow;
48 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
49 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
50 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowModInputBuilder;
51 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
52 import org.opendaylight.yangtools.yang.common.RpcError;
53 import org.opendaylight.yangtools.yang.common.RpcResult;
54 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
55 import org.slf4j.Logger;
56 import org.slf4j.LoggerFactory;
58 public class SalFlowServiceImpl implements SalFlowService {
59 private static final Logger LOG = LoggerFactory.getLogger(SalFlowServiceImpl.class);
60 private final MultiLayerFlowService<UpdateFlowOutput> flowUpdate;
61 private final MultiLayerFlowService<AddFlowOutput> flowAdd;
62 private final MultiLayerFlowService<RemoveFlowOutput> flowRemove;
63 private final SingleLayerFlowService<AddFlowOutput> flowAddMessage;
64 private final SingleLayerFlowService<UpdateFlowOutput> flowUpdateMessage;
65 private final SingleLayerFlowService<RemoveFlowOutput> flowRemoveMessage;
66 private final DeviceContext deviceContext;
68 public SalFlowServiceImpl(final RequestContextStack requestContextStack,
69 final DeviceContext deviceContext,
70 final ConvertorExecutor convertorExecutor) {
71 this.deviceContext = deviceContext;
72 flowRemove = new MultiLayerFlowService<>(requestContextStack,
74 RemoveFlowOutput.class,
76 flowAdd = new MultiLayerFlowService<>(requestContextStack,
80 flowUpdate = new MultiLayerFlowService<>(requestContextStack,
82 UpdateFlowOutput.class,
84 flowAddMessage = new SingleLayerFlowService<>(requestContextStack, deviceContext, AddFlowOutput.class);
85 flowUpdateMessage = new SingleLayerFlowService<>(requestContextStack, deviceContext, UpdateFlowOutput.class);
86 flowRemoveMessage = new SingleLayerFlowService<>(requestContextStack, deviceContext, RemoveFlowOutput.class);
90 public Future<RpcResult<AddFlowOutput>> addFlow(final AddFlowInput input) {
91 final FlowRegistryKey flowRegistryKey =
92 FlowRegistryKeyFactory.create(deviceContext.getDeviceInfo().getVersion(), input);
93 final ListenableFuture<RpcResult<AddFlowOutput>> future;
95 if (flowAddMessage.canUseSingleLayerSerialization()) {
96 future = flowAddMessage.handleServiceCall(input);
97 Futures.addCallback(future, new AddFlowCallback(input, flowRegistryKey));
99 future = flowAdd.processFlowModInputBuilders(flowAdd.toFlowModInputs(input));
100 Futures.addCallback(future, new AddFlowCallback(input, flowRegistryKey));
107 public Future<RpcResult<RemoveFlowOutput>> removeFlow(final RemoveFlowInput input) {
108 final ListenableFuture<RpcResult<RemoveFlowOutput>> future;
110 if (flowRemoveMessage.canUseSingleLayerSerialization()) {
111 future = flowRemoveMessage.handleServiceCall(input);
112 Futures.addCallback(future, new RemoveFlowCallback(input));
115 future = flowRemove.processFlowModInputBuilders(flowRemove.toFlowModInputs(input));
116 Futures.addCallback(future, new RemoveFlowCallback(input));
123 public Future<RpcResult<UpdateFlowOutput>> updateFlow(final UpdateFlowInput input) {
124 final UpdatedFlow updated = input.getUpdatedFlow();
125 final OriginalFlow original = input.getOriginalFlow();
127 final List<FlowModInputBuilder> allFlowMods = new ArrayList<>();
128 final List<FlowModInputBuilder> ofFlowModInputs;
130 ListenableFuture<RpcResult<UpdateFlowOutput>> future;
131 if (flowUpdateMessage.canUseSingleLayerSerialization()) {
133 if (!FlowCreatorUtil.canModifyFlow(original, updated, flowUpdateMessage.getVersion())) {
134 final SettableFuture<RpcResult<UpdateFlowOutput>> objectSettableFuture = SettableFuture.create();
136 final ListenableFuture<List<RpcResult<UpdateFlowOutput>>> listListenableFuture =
137 Futures.successfulAsList(flowUpdateMessage.handleServiceCall(input.getOriginalFlow()),
138 flowUpdateMessage.handleServiceCall(input.getUpdatedFlow()));
140 Futures.addCallback(listListenableFuture, new FutureCallback<List<RpcResult<UpdateFlowOutput>>>() {
142 public void onSuccess(final List<RpcResult<UpdateFlowOutput>> results) {
143 final ArrayList<RpcError> errors = new ArrayList();
144 for (RpcResult<UpdateFlowOutput> flowModResult : results) {
145 if (flowModResult == null) {
146 errors.add(RpcResultBuilder.newError(
147 RpcError.ErrorType.PROTOCOL, OFConstants.APPLICATION_TAG,
148 "unexpected flowMod result (null) occurred"));
149 } else if (!flowModResult.isSuccessful()) {
150 errors.addAll(flowModResult.getErrors());
154 final RpcResultBuilder<UpdateFlowOutput> rpcResultBuilder;
155 if (errors.isEmpty()) {
156 rpcResultBuilder = RpcResultBuilder.success();
158 rpcResultBuilder = RpcResultBuilder.<UpdateFlowOutput>failed().withRpcErrors(errors);
161 objectSettableFuture.set(rpcResultBuilder.build());
165 public void onFailure(final Throwable throwable) {
166 RpcResultBuilder<UpdateFlowOutput> rpcResultBuilder = RpcResultBuilder.failed();
167 objectSettableFuture.set(rpcResultBuilder.build());
171 future = objectSettableFuture;
173 future = flowUpdateMessage.handleServiceCall(input.getUpdatedFlow());
176 if (!FlowCreatorUtil.canModifyFlow(original, updated, flowUpdate.getVersion())) {
177 // We would need to remove original and add updated.
180 final RemoveFlowInputBuilder removeflow = new RemoveFlowInputBuilder(original);
181 final List<FlowModInputBuilder> ofFlowRemoveInput = flowUpdate.toFlowModInputs(removeflow.build());
182 // remove flow should be the first
183 allFlowMods.addAll(ofFlowRemoveInput);
184 final AddFlowInputBuilder addFlowInputBuilder = new AddFlowInputBuilder(updated);
185 ofFlowModInputs = flowUpdate.toFlowModInputs(addFlowInputBuilder.build());
187 ofFlowModInputs = flowUpdate.toFlowModInputs(updated);
190 allFlowMods.addAll(ofFlowModInputs);
192 future = flowUpdate.processFlowModInputBuilders(allFlowMods);
195 Futures.addCallback(future, new UpdateFlowCallback(input));
200 private static KeyedInstanceIdentifier<Flow, FlowKey> createFlowPath(
201 FlowDescriptor flowDescriptor,
202 KeyedInstanceIdentifier<Node, NodeKey> nodePath) {
203 return nodePath.augmentation(FlowCapableNode.class)
204 .child(Table.class, flowDescriptor.getTableKey())
205 .child(Flow.class, new FlowKey(flowDescriptor.getFlowId()));
208 private final class AddFlowCallback implements FutureCallback<RpcResult<AddFlowOutput>> {
209 private final AddFlowInput input;
210 private final FlowRegistryKey flowRegistryKey;
212 private AddFlowCallback(final AddFlowInput input,
213 final FlowRegistryKey flowRegistryKey) {
215 this.flowRegistryKey = flowRegistryKey;
219 public void onSuccess(final RpcResult<AddFlowOutput> rpcResult) {
220 if (rpcResult.isSuccessful()) {
221 final FlowDescriptor flowDescriptor;
223 if (Objects.nonNull(input.getFlowRef())) {
224 final FlowId flowId = input.getFlowRef().getValue().firstKeyOf(Flow.class, FlowKey.class).getId();
225 flowDescriptor = FlowDescriptorFactory.create(input.getTableId(), flowId);
226 deviceContext.getDeviceFlowRegistry().storeDescriptor(flowRegistryKey, flowDescriptor);
228 deviceContext.getDeviceFlowRegistry().store(flowRegistryKey);
229 flowDescriptor = deviceContext.getDeviceFlowRegistry().retrieveDescriptor(flowRegistryKey);
232 if (LOG.isDebugEnabled()) {
233 LOG.debug("Flow add with id={} finished without error", flowDescriptor.getFlowId().getValue());
236 if (LOG.isDebugEnabled()) {
237 LOG.debug("Flow add failed for flow={}, errors={}", input,
238 ErrorUtil.errorsToString(rpcResult.getErrors()));
244 public void onFailure(final Throwable throwable) {
245 LOG.warn("Service call for adding flow={} failed, reason: {}", input, throwable);
249 private final class RemoveFlowCallback implements FutureCallback<RpcResult<RemoveFlowOutput>> {
250 private final RemoveFlowInput input;
252 private RemoveFlowCallback(final RemoveFlowInput input) {
257 public void onSuccess(final RpcResult<RemoveFlowOutput> result) {
258 if (result.isSuccessful()) {
259 if (LOG.isDebugEnabled()) {
260 LOG.debug("Flow remove finished without error for flow={}", input);
262 FlowRegistryKey flowRegistryKey =
263 FlowRegistryKeyFactory.create(deviceContext.getDeviceInfo().getVersion(), input);
264 deviceContext.getDeviceFlowRegistry().addMark(flowRegistryKey);
266 if (LOG.isDebugEnabled()) {
267 LOG.debug("Flow remove failed for flow={}, errors={}", input,
268 ErrorUtil.errorsToString(result.getErrors()));
274 public void onFailure(final Throwable throwable) {
275 LOG.warn("Service call for removing flow={} failed, reason: {}", input, throwable);
279 private final class UpdateFlowCallback implements FutureCallback<RpcResult<UpdateFlowOutput>> {
280 private final UpdateFlowInput input;
282 private UpdateFlowCallback(UpdateFlowInput input) {
287 public void onSuccess(final RpcResult<UpdateFlowOutput> updateFlowOutputRpcResult) {
288 final DeviceFlowRegistry deviceFlowRegistry = deviceContext.getDeviceFlowRegistry();
290 final UpdatedFlow updated = input.getUpdatedFlow();
291 final OriginalFlow original = input.getOriginalFlow();
292 final FlowRegistryKey origFlowRegistryKey =
293 FlowRegistryKeyFactory.create(deviceContext.getDeviceInfo().getVersion(), original);
294 final FlowRegistryKey updatedFlowRegistryKey =
295 FlowRegistryKeyFactory.create(deviceContext.getDeviceInfo().getVersion(), updated);
296 final FlowDescriptor origFlowDescriptor = deviceFlowRegistry.retrieveDescriptor(origFlowRegistryKey);
298 final boolean isUpdate = Objects.nonNull(origFlowDescriptor);
299 final FlowDescriptor updatedFlowDescriptor;
301 if (Objects.nonNull(input.getFlowRef())) {
302 updatedFlowDescriptor =
303 FlowDescriptorFactory.create(updated.getTableId(),
304 input.getFlowRef().getValue().firstKeyOf(Flow.class).getId());
307 updatedFlowDescriptor = origFlowDescriptor;
309 deviceFlowRegistry.store(updatedFlowRegistryKey);
310 updatedFlowDescriptor = deviceFlowRegistry.retrieveDescriptor(updatedFlowRegistryKey);
315 deviceFlowRegistry.addMark(origFlowRegistryKey);
316 deviceFlowRegistry.storeDescriptor(updatedFlowRegistryKey, updatedFlowDescriptor);
321 public void onFailure(final Throwable throwable) {
322 LOG.warn("Service call for updating flow={} failed, reason: {}", input, throwable);