southbound/southbound-impl/src/main/java/org/opendaylight/ovsdb/southbound/OvsdbDataChangeListener.java

   1 package org.opendaylight.ovsdb.southbound;
   2
   3 import java.net.UnknownHostException;
   4 import java.util.HashSet;
   5 import java.util.Map;
   6 import java.util.Map.Entry;
   7 import java.util.Set;
   8
   9 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  10 import org.opendaylight.controller.md.sal.binding.api.DataChangeListener;
  11 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
  12 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent;
  13 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  14 import org.opendaylight.ovsdb.lib.OvsdbClient;
  15 import org.opendaylight.ovsdb.southbound.ovsdb.transact.BridgeOperationalState;
  16 import org.opendaylight.ovsdb.southbound.ovsdb.transact.DataChangesManagedByOvsdbNodeEvent;
  17 import org.opendaylight.ovsdb.southbound.ovsdb.transact.TransactCommandAggregator;
  18 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.rev150105.OvsdbBridgeAugmentation;
  19 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.rev150105.OvsdbNodeAugmentation;
  20 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
  21 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
  22 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
  23 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  24 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  25 import org.opendaylight.yangtools.yang.binding.DataObject;
  26 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  27 import org.slf4j.Logger;
  28 import org.slf4j.LoggerFactory;
  29
  30 import com.google.common.base.Preconditions;
  31 import com.google.common.base.Predicates;
  32 import com.google.common.collect.Maps;
  33
  34 public class OvsdbDataChangeListener implements DataChangeListener, AutoCloseable {
  35
  36     private ListenerRegistration<DataChangeListener> registration;
  37     private OvsdbConnectionManager cm;
  38     private DataBroker db;
  39     private static final Logger LOG = LoggerFactory.getLogger(OvsdbDataChangeListener.class);
  40
  41     OvsdbDataChangeListener(DataBroker db, OvsdbConnectionManager cm) {
  42         LOG.info("Registering OvsdbNodeDataChangeListener");
  43         this.cm = cm;
  44         this.db = db;
  45         InstanceIdentifier<Node> path = InstanceIdentifier
  46                 .create(NetworkTopology.class)
  47                 .child(Topology.class, new TopologyKey(SouthboundConstants.OVSDB_TOPOLOGY_ID))
  48                 .child(Node.class);
  49         registration =
  50                 db.registerDataChangeListener(LogicalDatastoreType.CONFIGURATION, path, this, DataChangeScope.SUBTREE);
  51
  52     }
  53
  54     @Override
  55     public void close() throws Exception {
  56         registration.close();
  57     }
  58
  59     @Override
  60     public void onDataChanged(
  61             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
  62         // Connect first if we have to:
  63         connect(changes);
  64
  65         // Second update connections if we have to
  66         updateConnections(changes);
  67
  68         // Then handle updates to the actual data
  69         updateData(changes);
  70
  71         // Finally disconnect if we need to
  72         disconnect(changes);
  73
  74     }
  75
  76     private void updateData(
  77             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
  78         for (OvsdbConnectionInstance connectionInstance : connectionInstancesFromChanges(changes)) {
  79             connectionInstance.transact(new TransactCommandAggregator(
  80                     new BridgeOperationalState(db, changes),
  81                     new DataChangesManagedByOvsdbNodeEvent(
  82                             SouthboundMapper.createInstanceIdentifier(connectionInstance.getKey()),
  83                             changes)));
  84         }
  85     }
  86
  87     private void disconnect(
  88             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
  89         Map<InstanceIdentifier<?>, DataObject> originalDataObject = changes.getOriginalData();
  90         Set<InstanceIdentifier<?>> iiD = changes.getRemovedPaths();
  91         for (InstanceIdentifier instanceIdentifier : iiD) {
  92             if (originalDataObject.get(instanceIdentifier) instanceof OvsdbNodeAugmentation) {
  93                 try {
  94                     cm.disconnect((OvsdbNodeAugmentation) originalDataObject.get(instanceIdentifier));
  95                 } catch (UnknownHostException e) {
  96                     LOG.warn("Failed to disconnect ovsdbNode", e);
  97                 }
  98             }
  99         }
 100     }
 101
 102     private void updateConnections(
 103             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
 104         for (Entry<InstanceIdentifier<?>, DataObject> updated : changes.getUpdatedData().entrySet()) {
 105             if (updated.getValue() instanceof OvsdbNodeAugmentation) {
 106                 OvsdbNodeAugmentation value = (OvsdbNodeAugmentation) updated.getValue();
 107                 OvsdbClient client = cm.getClient(value.getConnectionInfo());
 108                 if (client == null) {
 109                     for (Entry<InstanceIdentifier<?>, DataObject> original : changes.getOriginalData().entrySet()) {
 110                         if (original.getValue() instanceof OvsdbNodeAugmentation) {
 111                             try {
 112                                 cm.disconnect((OvsdbNodeAugmentation) original.getValue());
 113                                 cm.connect(value);
 114                             } catch (UnknownHostException e) {
 115                                 LOG.warn("Failed to disconnect to ovsdbNode", e);
 116                             }
 117                         }
 118                     }
 119                 }
 120             }
 121         }
 122     }
 123
 124     private void connect(
 125             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
 126         for (Entry<InstanceIdentifier<?>, DataObject> created : changes.getCreatedData().entrySet()) {
 127             // TODO validate we have the correct kind of InstanceIdentifier
 128             if (created.getValue() instanceof OvsdbNodeAugmentation) {
 129                 try {
 130                     cm.connect((OvsdbNodeAugmentation) created.getValue());
 131                 } catch (UnknownHostException e) {
 132                     LOG.warn("Failed to connect to ovsdbNode", e);
 133                 }
 134             }
 135         }
 136     }
 137
 138     public Set<OvsdbConnectionInstance> connectionInstancesFromChanges(
 139             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changes) {
 140         Set<OvsdbConnectionInstance> result = new HashSet<OvsdbConnectionInstance>();
 141         result.addAll(connectionInstancesFromMap(changes.getCreatedData()));
 142         result.addAll(connectionInstancesFromMap(changes.getUpdatedData()));
 143         result.addAll(connectionInstancesFromMap(
 144                 Maps.filterKeys(changes.getOriginalData(), Predicates.in(changes.getRemovedPaths()))));
 145         return result;
 146     }
 147
 148     public Set<OvsdbConnectionInstance> connectionInstancesFromMap(Map<InstanceIdentifier<?>, DataObject> map) {
 149         Preconditions.checkNotNull(map);
 150         Set<OvsdbConnectionInstance> result = new HashSet<OvsdbConnectionInstance>();
 151         for ( Entry<InstanceIdentifier<?>, DataObject> created : map.entrySet()) {
 152             if (created.getValue() instanceof Node) {
 153                 LOG.debug("Received request to create {}",created.getValue());
 154                 OvsdbBridgeAugmentation bridge =
 155                         ((Node)created.getValue()).getAugmentation(OvsdbBridgeAugmentation.class);
 156                 if (bridge != null) {
 157                     OvsdbConnectionInstance client = cm.getConnectionInstance(bridge);
 158                     if (client != null) {
 159                         LOG.debug("Found client for {}", created.getValue());
 160                         result.add(client);
 161                     } else {
 162                         LOG.debug("Did not find client for {}",created.getValue());
 163                     }
 164                 } else {
 165                     OvsdbNodeAugmentation ovsNode =
 166                             ((Node)created.getValue()).getAugmentation(OvsdbNodeAugmentation.class);
 167                     if (ovsNode != null && ovsNode.getConnectionInfo() != null) {
 168                         OvsdbConnectionInstance client = cm.getConnectionInstance(ovsNode.getConnectionInfo());
 169                         if (client != null) {
 170                             LOG.debug("Found client for {}", created.getValue());
 171                             result.add(client);
 172                         } else {
 173                             LOG.debug("Did not find client for {}",created.getValue());
 174                         }
 175                     }
 176                 }
 177             }
 178         }
 179         return result;
 180     }
 181
 182 }