2 * Copyright (c) 2014 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.controller.sal.compatibility;
10 import java.util.EnumSet;
12 import java.util.UUID;
13 import java.util.concurrent.ConcurrentHashMap;
14 import java.util.concurrent.ExecutionException;
15 import java.util.concurrent.Future;
17 import org.opendaylight.controller.clustering.services.CacheConfigException;
18 import org.opendaylight.controller.clustering.services.CacheExistException;
19 import org.opendaylight.controller.clustering.services.IClusterGlobalServices;
20 import org.opendaylight.controller.clustering.services.IClusterServices.cacheMode;
21 import org.opendaylight.controller.md.sal.common.api.TransactionStatus;
22 import org.opendaylight.controller.sal.binding.api.data.DataBrokerService;
23 import org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction;
24 import org.opendaylight.controller.sal.core.ConstructionException;
25 import org.opendaylight.controller.sal.core.Node;
26 import org.opendaylight.controller.sal.flowprogrammer.Flow;
27 import org.opendaylight.controller.sal.flowprogrammer.IPluginInFlowProgrammerService;
28 import org.opendaylight.controller.sal.flowprogrammer.IPluginOutFlowProgrammerService;
29 import org.opendaylight.controller.sal.utils.Status;
30 import org.opendaylight.controller.sal.utils.StatusCode;
31 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
32 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
33 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
34 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
35 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowAdded;
36 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved;
37 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowUpdated;
38 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.NodeErrorNotification;
39 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.NodeExperimenterErrorNotification;
40 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.SalFlowListener;
41 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.SalFlowService;
42 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.SwitchFlowRemoved;
43 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
44 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
45 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
46 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
47 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
48 import org.opendaylight.yangtools.yang.common.RpcResult;
49 import org.slf4j.Logger;
50 import org.slf4j.LoggerFactory;
52 public class FlowProgrammerAdapter implements IPluginInFlowProgrammerService, SalFlowListener {
53 private final static Logger LOG = LoggerFactory.getLogger(FlowProgrammerAdapter.class);
55 // Note: clustering services manipulate this
56 private final Map<Flow, UUID> flowToFlowId = new ConcurrentHashMap<Flow, UUID>();
57 private final static String CACHE_NAME = "flowprogrammeradapter.flowtoid";
59 // These are injected via Apache DM (see ComponentActivator)
60 private IPluginOutFlowProgrammerService flowProgrammerPublisher;
61 private IClusterGlobalServices clusterGlobalServices;
62 private DataBrokerService dataBrokerService;
63 private SalFlowService delegate;
65 public SalFlowService getDelegate() {
69 public void setDelegate(final SalFlowService delegate) {
70 this.delegate = delegate;
73 public DataBrokerService getDataBrokerService() {
74 return this.dataBrokerService;
77 public void setDataBrokerService(final DataBrokerService dataBrokerService) {
78 this.dataBrokerService = dataBrokerService;
81 public IPluginOutFlowProgrammerService getFlowProgrammerPublisher() {
82 return this.flowProgrammerPublisher;
85 public void setFlowProgrammerPublisher(final IPluginOutFlowProgrammerService flowProgrammerPublisher) {
86 this.flowProgrammerPublisher = flowProgrammerPublisher;
89 public IClusterGlobalServices getClusterGlobalServices() {
90 return this.clusterGlobalServices;
93 public void setClusterGlobalServices(final IClusterGlobalServices clusterGlobalServices) {
94 this.clusterGlobalServices = clusterGlobalServices;
98 public Status addFlow(final Node node, final Flow flow) {
99 return toFutureStatus(internalAddFlowAsync(node, flow, 0));
103 public Status modifyFlow(final Node node, final Flow oldFlow, final Flow newFlow) {
104 return toFutureStatus(internalModifyFlowAsync(node, oldFlow, newFlow, 0));
108 public Status removeFlow(final Node node, final Flow flow) {
109 return toFutureStatus(internalRemoveFlowAsync(node, flow, 0));
113 public Status addFlowAsync(final Node node, final Flow flow, final long rid) {
114 // FIXME is this correct? What if the future fails?
115 this.internalAddFlowAsync(node, flow, rid);
116 return FlowProgrammerAdapter.toStatus(true);
120 public Status modifyFlowAsync(final Node node, final Flow oldFlow, final Flow newFlow, final long rid) {
121 // FIXME is this correct? What if the future fails?
122 this.internalModifyFlowAsync(node, oldFlow, newFlow, rid);
123 return FlowProgrammerAdapter.toStatus(true);
127 public Status removeFlowAsync(final Node node, final Flow flow, final long rid) {
128 // FIXME is this correct? What if the future fails?
129 this.internalRemoveFlowAsync(node, flow, rid);
130 return FlowProgrammerAdapter.toStatus(true);
134 public Status removeAllFlows(final Node node) {
135 // FIXME: unfinished?
136 return new Status(StatusCode.SUCCESS);
140 public Status syncSendBarrierMessage(final Node node) {
141 // FIXME: unfinished?
146 public Status asyncSendBarrierMessage(final Node node) {
147 // FIXME: unfinished?
151 private static Status toStatus(final boolean successful) {
152 return new Status(successful ? StatusCode.SUCCESS : StatusCode.INTERNALERROR);
155 public static Status toStatus(final RpcResult<? extends Object> result) {
156 return toStatus(result.isSuccessful());
160 public void onFlowAdded(final FlowAdded notification) {
161 // FIXME: unfinished?
165 public void onFlowRemoved(final FlowRemoved notification) {
166 // notified upon remove flow rpc successfully invoked
167 if (notification == null) {
171 final NodeRef node = notification.getNode();
173 LOG.debug("Notification {} has not node, ignoring it", notification);
179 adNode = NodeMapping.toADNode(notification.getNode());
180 } catch (ConstructionException e) {
181 LOG.warn("Failed to construct AD node for {}, ignoring notification", node, e);
184 flowProgrammerPublisher.flowRemoved(adNode, ToSalConversionsUtils.toFlow(notification, adNode));
188 public void onFlowUpdated(final FlowUpdated notification) {
189 // FIXME: unfinished?
193 public void onSwitchFlowRemoved(final SwitchFlowRemoved notification) {
194 // notified upon remove flow message from device arrives
195 if (notification == null) {
199 final NodeRef node = notification.getNode();
201 LOG.debug("Notification {} has not node, ignoring it", notification);
207 adNode = NodeMapping.toADNode(notification.getNode());
208 } catch (ConstructionException e) {
209 LOG.warn("Failed to construct AD node for {}, ignoring notification", node, e);
212 flowProgrammerPublisher.flowRemoved(adNode, ToSalConversionsUtils.toFlow(notification, adNode));
216 public void onNodeErrorNotification(final NodeErrorNotification notification) {
217 // FIXME: unfinished?
221 public void onNodeExperimenterErrorNotification(final NodeExperimenterErrorNotification notification) {
222 // FIXME: unfinished?
225 private static final InstanceIdentifier<org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow> flowPath(
226 final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow flow, final NodeKey nodeKey) {
227 return InstanceIdentifier.builder(Nodes.class)
228 .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node.class, nodeKey)
229 .augmentation(FlowCapableNode.class)
230 .child(Table.class, new TableKey(flow.getTableId()))
231 .child(org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow.class, new FlowKey(flow.getId()))
235 private Future<RpcResult<TransactionStatus>> writeFlowAsync(final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow flow, final NodeKey nodeKey) {
236 final DataModificationTransaction modification = this.dataBrokerService.beginTransaction();
237 modification.putConfigurationData(flowPath(flow, nodeKey), flow);
238 return modification.commit();
241 private Future<RpcResult<TransactionStatus>> internalAddFlowAsync(final Node node, final Flow flow, final long rid) {
242 final Map<Flow,UUID> cache = this.getCache();
243 UUID flowId = cache.get(flow);
244 if (flowId != null) {
245 this.removeFlow(node, flow);
248 flowId = UUID.randomUUID();
249 cache.put(flow, flowId);
250 return this.writeFlowAsync(MDFlowMapping.toMDFlow(flow, flowId.toString()), new NodeKey(
251 new NodeId(NodeMapping.OPENFLOW_ID_PREFIX + node.getID())));
254 private Future<RpcResult<TransactionStatus>> internalModifyFlowAsync(final Node node, final Flow oldFlow, final Flow newFlow, final long rid) {
255 final Map<Flow,UUID> cache = this.getCache();
257 UUID flowId = cache.remove(oldFlow);
258 if (flowId == null) {
259 flowId = UUID.randomUUID();
260 cache.put(oldFlow, flowId);
261 LOG.warn("Could not find flow {} in cache, assigned new ID {}", oldFlow.hashCode(), flowId);
264 cache.put(newFlow, flowId);
265 return this.writeFlowAsync(MDFlowMapping.toMDFlow(newFlow, flowId.toString()), new NodeKey(
266 new NodeId(NodeMapping.OPENFLOW_ID_PREFIX + node.getID())));
269 private Future<RpcResult<TransactionStatus>> internalRemoveFlowAsync(final Node node, final Flow adflow, final long rid) {
270 final Map<Flow,UUID> cache = this.getCache();
272 final UUID flowId = cache.remove(adflow);
273 if (flowId == null) {
274 LOG.warn("Could not find flow {} in cache, nothing to do", adflow.hashCode());
278 final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow flow = MDFlowMapping.toMDFlow(adflow, flowId.toString());
279 final DataModificationTransaction modification = this.dataBrokerService.beginTransaction();
280 modification.removeConfigurationData(flowPath(flow, new NodeKey(
281 new NodeId(NodeMapping.OPENFLOW_ID_PREFIX + node.getID()))));
282 return modification.commit();
285 private static Status toFutureStatus(final Future<RpcResult<TransactionStatus>> future) {
286 if (future == null) {
288 return FlowProgrammerAdapter.toStatus(true);
292 final RpcResult<TransactionStatus> result = future.get();
293 return FlowProgrammerAdapter.toStatus(result);
294 } catch (final InterruptedException e) {
295 FlowProgrammerAdapter.LOG.error("Interrupted while processing flow", e);
296 } catch (ExecutionException e) {
297 FlowProgrammerAdapter.LOG.error("Failed to process flow", e);
300 return new Status(StatusCode.INTERNALERROR);
303 @SuppressWarnings("unchecked")
304 private Map<Flow,UUID> getCache() {
305 final IClusterGlobalServices cgs = getClusterGlobalServices();
307 return new ConcurrentHashMap<Flow, UUID>();
310 Map<Flow, UUID> cache = (Map<Flow, UUID>) cgs.getCache(FlowProgrammerAdapter.CACHE_NAME);
316 return (Map<Flow, UUID>) cgs.createCache(CACHE_NAME, EnumSet.of(cacheMode.TRANSACTIONAL));
317 } catch (CacheExistException e) {
318 return (Map<Flow, UUID>) cgs.getCache(CACHE_NAME);
319 } catch (CacheConfigException e) {
320 throw new IllegalStateException("Unexpected cache configuration problem", e);