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 javax.annotation.Nullable;
20 import org.opendaylight.openflowplugin.api.OFConstants;
21 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
22 import org.opendaylight.openflowplugin.api.openflow.device.RequestContextStack;
23 import org.opendaylight.openflowplugin.api.openflow.registry.flow.DeviceFlowRegistry;
24 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowDescriptor;
25 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowRegistryKey;
26 import org.opendaylight.openflowplugin.api.openflow.rpc.ItemLifeCycleSource;
27 import org.opendaylight.openflowplugin.api.openflow.rpc.listener.ItemLifecycleListener;
28 import org.opendaylight.openflowplugin.impl.registry.flow.FlowDescriptorFactory;
29 import org.opendaylight.openflowplugin.impl.registry.flow.FlowRegistryKeyFactory;
30 import org.opendaylight.openflowplugin.impl.services.singlelayer.SingleLayerFlowService;
31 import org.opendaylight.openflowplugin.impl.services.multilayer.MultiLayerFlowService;
32 import org.opendaylight.openflowplugin.impl.util.ErrorUtil;
33 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
34 import org.opendaylight.openflowplugin.openflow.md.util.FlowCreatorUtil;
35 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
36 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
37 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
38 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
39 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowBuilder;
40 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
41 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowInput;
42 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowInputBuilder;
43 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowOutput;
44 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.RemoveFlowInput;
45 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.RemoveFlowInputBuilder;
46 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.RemoveFlowOutput;
47 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.SalFlowService;
48 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.UpdateFlowInput;
49 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.UpdateFlowOutput;
50 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.flow.update.OriginalFlow;
51 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.flow.update.UpdatedFlow;
52 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
53 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
54 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowModInputBuilder;
55 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
56 import org.opendaylight.yangtools.yang.common.RpcError;
57 import org.opendaylight.yangtools.yang.common.RpcResult;
58 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
59 import org.slf4j.Logger;
60 import org.slf4j.LoggerFactory;
62 public class SalFlowServiceImpl implements SalFlowService, ItemLifeCycleSource {
63 private static final Logger LOG = LoggerFactory.getLogger(SalFlowServiceImpl.class);
64 private final MultiLayerFlowService<UpdateFlowOutput> flowUpdate;
65 private final MultiLayerFlowService<AddFlowOutput> flowAdd;
66 private final MultiLayerFlowService<RemoveFlowOutput> flowRemove;
67 private final SingleLayerFlowService<AddFlowOutput> flowAddMessage;
68 private final SingleLayerFlowService<UpdateFlowOutput> flowUpdateMessage;
69 private final SingleLayerFlowService<RemoveFlowOutput> flowRemoveMessage;
70 private final DeviceContext deviceContext;
71 private ItemLifecycleListener itemLifecycleListener;
73 public SalFlowServiceImpl(final RequestContextStack requestContextStack, final DeviceContext deviceContext, final ConvertorExecutor convertorExecutor) {
74 this.deviceContext = deviceContext;
75 flowRemove = new MultiLayerFlowService<>(requestContextStack, deviceContext, RemoveFlowOutput.class, convertorExecutor);
76 flowAdd = new MultiLayerFlowService<>(requestContextStack, deviceContext, AddFlowOutput.class, convertorExecutor);
77 flowUpdate = new MultiLayerFlowService<>(requestContextStack, deviceContext, UpdateFlowOutput.class, convertorExecutor);
78 flowAddMessage = new SingleLayerFlowService<>(requestContextStack, deviceContext, AddFlowOutput.class);
79 flowUpdateMessage = new SingleLayerFlowService<>(requestContextStack, deviceContext, UpdateFlowOutput.class);
80 flowRemoveMessage= new SingleLayerFlowService<>(requestContextStack, deviceContext, RemoveFlowOutput.class);
84 public void setItemLifecycleListener(@Nullable ItemLifecycleListener itemLifecycleListener) {
85 this.itemLifecycleListener = itemLifecycleListener;
89 public Future<RpcResult<AddFlowOutput>> addFlow(final AddFlowInput input) {
90 final FlowRegistryKey flowRegistryKey = FlowRegistryKeyFactory.create(deviceContext.getDeviceInfo().getVersion(), input);
91 final ListenableFuture<RpcResult<AddFlowOutput>> future;
93 if (flowAddMessage.canUseSingleLayerSerialization()) {
94 future = flowAddMessage.handleServiceCall(input);
95 Futures.addCallback(future, new AddFlowCallback(input, flowRegistryKey));
97 future = flowAdd.processFlowModInputBuilders(flowAdd.toFlowModInputs(input));
98 Futures.addCallback(future, new AddFlowCallback(input, flowRegistryKey));
105 public Future<RpcResult<RemoveFlowOutput>> removeFlow(final RemoveFlowInput input) {
106 final ListenableFuture<RpcResult<RemoveFlowOutput>> future;
108 if (flowRemoveMessage.canUseSingleLayerSerialization()) {
109 future = flowRemoveMessage.handleServiceCall(input);
110 Futures.addCallback(future, new RemoveFlowCallback(input));
113 future = flowRemove.processFlowModInputBuilders(flowRemove.toFlowModInputs(input));
114 Futures.addCallback(future, new RemoveFlowCallback(input));
121 public Future<RpcResult<UpdateFlowOutput>> updateFlow(final UpdateFlowInput input) {
122 final UpdatedFlow updated = input.getUpdatedFlow();
123 final OriginalFlow original = input.getOriginalFlow();
125 final List<FlowModInputBuilder> allFlowMods = new ArrayList<>();
126 final List<FlowModInputBuilder> ofFlowModInputs;
128 ListenableFuture<RpcResult<UpdateFlowOutput>> future;
129 if (flowUpdateMessage.canUseSingleLayerSerialization()) {
131 if (!FlowCreatorUtil.canModifyFlow(original, updated, flowUpdateMessage.getVersion())) {
132 final SettableFuture<RpcResult<UpdateFlowOutput>> objectSettableFuture = SettableFuture.create();
134 final ListenableFuture<List<RpcResult<UpdateFlowOutput>>> listListenableFuture = Futures.successfulAsList(
135 flowUpdateMessage.handleServiceCall(input.getOriginalFlow()),
136 flowUpdateMessage.handleServiceCall(input.getUpdatedFlow()));
138 Futures.addCallback(listListenableFuture, new FutureCallback<List<RpcResult<UpdateFlowOutput>>>() {
140 public void onSuccess(final List<RpcResult<UpdateFlowOutput>> results) {
141 final ArrayList<RpcError> errors = new ArrayList();
142 for (RpcResult<UpdateFlowOutput> flowModResult : results) {
143 if (flowModResult == null) {
144 errors.add(RpcResultBuilder.newError(
145 RpcError.ErrorType.PROTOCOL, OFConstants.APPLICATION_TAG,
146 "unexpected flowMod result (null) occurred"));
147 } else if (!flowModResult.isSuccessful()) {
148 errors.addAll(flowModResult.getErrors());
152 final RpcResultBuilder<UpdateFlowOutput> rpcResultBuilder;
153 if (errors.isEmpty()) {
154 rpcResultBuilder = RpcResultBuilder.success();
156 rpcResultBuilder = RpcResultBuilder.<UpdateFlowOutput>failed().withRpcErrors(errors);
159 objectSettableFuture.set(rpcResultBuilder.build());
163 public void onFailure(final Throwable t) {
164 RpcResultBuilder<UpdateFlowOutput> rpcResultBuilder = RpcResultBuilder.failed();
165 objectSettableFuture.set(rpcResultBuilder.build());
169 future = objectSettableFuture;
171 future = flowUpdateMessage.handleServiceCall(input.getUpdatedFlow());
174 if (!FlowCreatorUtil.canModifyFlow(original, updated, flowUpdate.getVersion())) {
175 // We would need to remove original and add updated.
178 final RemoveFlowInputBuilder removeflow = new RemoveFlowInputBuilder(original);
179 final List<FlowModInputBuilder> ofFlowRemoveInput = flowUpdate.toFlowModInputs(removeflow.build());
180 // remove flow should be the first
181 allFlowMods.addAll(ofFlowRemoveInput);
182 final AddFlowInputBuilder addFlowInputBuilder = new AddFlowInputBuilder(updated);
183 ofFlowModInputs = flowUpdate.toFlowModInputs(addFlowInputBuilder.build());
185 ofFlowModInputs = flowUpdate.toFlowModInputs(updated);
188 allFlowMods.addAll(ofFlowModInputs);
190 future = flowUpdate.processFlowModInputBuilders(allFlowMods);
193 Futures.addCallback(future, new UpdateFlowCallback(input));
198 private static KeyedInstanceIdentifier<Flow, FlowKey> createFlowPath(FlowDescriptor flowDescriptor,
199 KeyedInstanceIdentifier<Node, NodeKey> nodePath) {
200 return nodePath.augmentation(FlowCapableNode.class)
201 .child(Table.class, flowDescriptor.getTableKey())
202 .child(Flow.class, new FlowKey(flowDescriptor.getFlowId()));
205 private class AddFlowCallback implements FutureCallback<RpcResult<AddFlowOutput>> {
206 private final AddFlowInput input;
207 private final FlowRegistryKey flowRegistryKey;
209 private AddFlowCallback(final AddFlowInput input,
210 final FlowRegistryKey flowRegistryKey) {
212 this.flowRegistryKey = flowRegistryKey;
216 public void onSuccess(final RpcResult<AddFlowOutput> rpcResult) {
217 if (rpcResult.isSuccessful()) {
218 final FlowDescriptor flowDescriptor;
220 if (Objects.nonNull(input.getFlowRef())) {
221 final FlowId flowId = input.getFlowRef().getValue().firstKeyOf(Flow.class, FlowKey.class).getId();
222 flowDescriptor = FlowDescriptorFactory.create(input.getTableId(), flowId);
223 deviceContext.getDeviceFlowRegistry().store(flowRegistryKey, flowDescriptor);
225 final FlowId flowId = deviceContext.getDeviceFlowRegistry().storeIfNecessary(flowRegistryKey);
226 flowDescriptor = FlowDescriptorFactory.create(input.getTableId(), flowId);
229 if (LOG.isDebugEnabled()) {
230 LOG.debug("Flow add with id={} finished without error", flowDescriptor.getFlowId().getValue());
233 if (itemLifecycleListener != null) {
234 KeyedInstanceIdentifier<Flow, FlowKey> flowPath = createFlowPath(flowDescriptor,
235 deviceContext.getDeviceInfo().getNodeInstanceIdentifier());
236 final FlowBuilder flowBuilder = new FlowBuilder(input).setId(flowDescriptor.getFlowId());
237 itemLifecycleListener.onAdded(flowPath, flowBuilder.build());
240 if (LOG.isDebugEnabled()) {
241 LOG.debug("Flow add failed for flow={}, errors={}", input,
242 ErrorUtil.errorsToString(rpcResult.getErrors()));
248 public void onFailure(final Throwable throwable) {
249 LOG.warn("Service call for adding flow={} failed, reason: {}", input, throwable);
253 private class RemoveFlowCallback implements FutureCallback<RpcResult<RemoveFlowOutput>> {
254 private final RemoveFlowInput input;
256 private RemoveFlowCallback(final RemoveFlowInput input) {
261 public void onSuccess(final RpcResult<RemoveFlowOutput> result) {
262 if (result.isSuccessful()) {
263 if (LOG.isDebugEnabled()) {
264 LOG.debug("Flow remove finished without error for flow={}", input);
266 FlowRegistryKey flowRegistryKey = FlowRegistryKeyFactory.create(deviceContext.getDeviceInfo().getVersion(), input);
267 deviceContext.getDeviceFlowRegistry().removeDescriptor(flowRegistryKey);
269 if (itemLifecycleListener != null) {
270 final FlowDescriptor flowDescriptor =
271 deviceContext.getDeviceFlowRegistry().retrieveIdForFlow(flowRegistryKey);
272 if (flowDescriptor != null) {
273 KeyedInstanceIdentifier<Flow, FlowKey> flowPath = createFlowPath(flowDescriptor,
274 deviceContext.getDeviceInfo().getNodeInstanceIdentifier());
275 itemLifecycleListener.onRemoved(flowPath);
279 if (LOG.isDebugEnabled()) {
280 LOG.debug("Flow remove failed for flow={}, errors={}", input,
281 ErrorUtil.errorsToString(result.getErrors()));
287 public void onFailure(final Throwable throwable) {
288 LOG.warn("Service call for removing flow={} failed, reason: {}", input, throwable);
292 private class UpdateFlowCallback implements FutureCallback<RpcResult<UpdateFlowOutput>> {
293 private final UpdateFlowInput input;
295 private UpdateFlowCallback(UpdateFlowInput input) {
300 public void onSuccess(final RpcResult<UpdateFlowOutput> o) {
301 final DeviceFlowRegistry deviceFlowRegistry = deviceContext.getDeviceFlowRegistry();
303 final UpdatedFlow updated = input.getUpdatedFlow();
304 final OriginalFlow original = input.getOriginalFlow();
305 final FlowRegistryKey origFlowRegistryKey = FlowRegistryKeyFactory.create(deviceContext.getDeviceInfo().getVersion(), original);
306 final FlowRegistryKey updatedFlowRegistryKey = FlowRegistryKeyFactory.create(deviceContext.getDeviceInfo().getVersion(), updated);
307 final FlowDescriptor origFlowDescriptor = deviceFlowRegistry.retrieveIdForFlow(origFlowRegistryKey);
309 final boolean isUpdate = Objects.nonNull(origFlowDescriptor);
310 final FlowId fLowId = Objects.nonNull(input.getFlowRef())
311 ? input.getFlowRef().getValue().firstKeyOf(Flow.class).getId()
312 : isUpdate ? origFlowDescriptor.getFlowId() : deviceFlowRegistry.storeIfNecessary(updatedFlowRegistryKey);
313 final FlowDescriptor updatedFlowDescriptor = FlowDescriptorFactory.create(updated.getTableId(), fLowId);
315 deviceFlowRegistry.removeDescriptor(origFlowRegistryKey);
316 deviceFlowRegistry.store(updatedFlowRegistryKey, updatedFlowDescriptor);
319 if (itemLifecycleListener != null) {
320 final KeyedInstanceIdentifier<Flow, FlowKey> flowPath =
322 updatedFlowDescriptor,
323 deviceContext.getDeviceInfo().getNodeInstanceIdentifier());
325 final Flow flow = new FlowBuilder(updated)
326 .setId(updatedFlowDescriptor.getFlowId())
329 if (Objects.nonNull(origFlowDescriptor)) {
330 itemLifecycleListener.onUpdated(flowPath, flow);
332 itemLifecycleListener.onAdded(flowPath, flow);
338 public void onFailure(final Throwable throwable) {
339 LOG.warn("Service call for updating flow={} failed, reason: {}", input, throwable);